杰瑞科技汇

Python如何调用REST API接口?

核心概念

REST API 通常通过 HTTP 协议进行通信,使用不同的请求方法来操作资源:

Python如何调用REST API接口?-图1
(图片来源网络,侵删)
  • GET: 获取资源(获取一个用户的信息)。
  • POST: 创建新资源(创建一个新用户)。
  • PUT / PATCH: 更新现有资源(更新用户信息)。
  • DELETE: 删除资源(删除一个用户)。

API 请求通常包含:

  1. URL: API 的端点地址。
  2. HTTP 方法: 上述的 GET, POST 等。
  3. Headers: 请求头,用于传递元数据,如 Content-Type类型)、Authorization(认证信息)等。
  4. Body (请求体): 对于 POST/PUT/PATCH 请求,通常需要发送数据,如 JSON 格式的数据。

使用 requests 库(推荐)

requests 是 Python 中最流行、最易用的 HTTP 库,它极大地简化了发送 HTTP 请求的过程。

安装 requests

如果你还没有安装,可以通过 pip 安装:

pip install requests

发送 GET 请求

GET 请求用于从服务器获取数据。

Python如何调用REST API接口?-图2
(图片来源网络,侵删)

示例:从 JSONPlaceholder 获取一个用户的 JSON 数据

import requests
import json # 用于美化打印 JSON
# API 端点
url = "https://jsonplaceholder.typicode.com/users/1"
try:
    # 发送 GET 请求
    response = requests.get(url)
    # 检查请求是否成功 (状态码 200-299 表示成功)
    response.raise_for_status()  # 如果请求失败 (404, 500), 这行代码会抛出异常
    # 获取响应内容
    # response.text 是原始字符串
    # response.json() 会自动将响应内容解析为 Python 字典或列表 (如果响应是 JSON 格式)
    user_data = response.json()
    # 打印获取到的数据
    print("成功获取用户数据:")
    print(json.dumps(user_data, indent=4, ensure_ascii=False))
except requests.exceptions.HTTPError as http_err:
    print(f"HTTP 错误发生: {http_err}")
except requests.exceptions.ConnectionError as conn_err:
    print(f"连接错误发生: {conn_err}")
except requests.exceptions.Timeout as timeout_err:
    print(f"请求超时: {timeout_err}")
except requests.exceptions.RequestException as err:
    print(f"发生未知错误: {err}")

发送 POST 请求

POST 请求用于向服务器提交数据,通常用于创建新资源。

示例:向 JSONPlaceholder 创建一个新的帖子

import requests
import json
# API 端点
url = "https://jsonplaceholder.typicode.com/posts"
# 要创建的数据 (一个 Python 字典)
new_post_data = {: 'Python 学习笔记',
    'body': '今天学习了如何使用 requests 库调用 REST API。',
    'userId': 1
}
try:
    # 发送 POST 请求
    # json=new_post_data 会自动将字典转换为 JSON 格式,
    # 并设置正确的 Content-Type header 为 application/json
    response = requests.post(url, json=new_post_data)
    # 检查请求是否成功
    response.raise_for_status()
    # 获取服务器返回的创建后的数据 (通常包含新资源的 ID)
    created_post = response.json()
    print("成功创建新帖子:")
    print(json.dumps(created_post, indent=4, ensure_ascii=False))
except requests.exceptions.RequestException as err:
    print(f"发生错误: {err}")

处理 Headers 和 Query Parameters

很多 API 需要认证信息(如 API Key)或通过查询参数来过滤结果。

Python如何调用REST API接口?-图3
(图片来源网络,侵删)

示例:带 Header 和查询参数的 GET 请求

import requests
import json
# API 端点
url = "https://api.github.com/users"
# 查询参数: 搜索 "python" 关键词,并限制结果数量
params = {
    'q': 'python',
    'per_page': 5
}
# 请求头,包含自定义的 User-Agent 和 API Key (示例)
# 很多 API 需要在 Header 中传递 API Key
headers = {
    'User-Agent': 'My-Python-App/1.0',
    # 'Authorization': 'Bearer YOUR_API_KEY' # 替换成你的真实 API Key
}
try:
    # 发送 GET 请求,传递 params 和 headers
    response = requests.get(url, params=params, headers=headers)
    response.raise_for_status()
    users = response.json()
    print("成功搜索到 GitHub 用户:")
    for user in users:
        print(f"- {user['login']} (ID: {user['id']})")
except requests.exceptions.RequestException as err:
    print(f"发生错误: {err}")

使用 urllib 库(标准库)

urllib 是 Python 的标准库,无需安装,但它比 requests 更底层,使用起来更繁琐,不推荐用于复杂的 API 调用,了解它有助于理解底层原理。

import urllib.request
import urllib.parse
import json
url = "https://jsonplaceholder.typicode.com/users/1"
try:
    # 创建请求对象
    req = urllib.request.Request(url)
    # 发送请求并获取响应
    with urllib.request.urlopen(req) as response:
        # 读取响应内容
        response_data = response.read().decode('utf-8')
        # 解析 JSON
        user_data = json.loads(response_data)
        print("使用 urllib 获取用户数据:")
        print(json.dumps(user_data, indent=4, ensure_ascii=False))
except urllib.error.HTTPError as err:
    print(f"HTTP 错误: {err.code} {err.reason}")
except urllib.error.URLError as err:
    print(f"URL 错误: {err.reason}")
except Exception as err:
    print(f"发生错误: {err}")

POST 请求示例(urllib

import urllib.request
import urllib.parse
import json
url = "https://jsonplaceholder.typicode.com/posts"
data = {: 'urllib 测试',
    'body': '使用标准库发送 POST 请求',
    'userId': 1
}
# 将字典编码为 JSON 字符串,并编码为 bytes
post_data = json.dumps(data).encode('utf-8')
# 创建请求对象,需要手动指定 method 和 headers
req = urllib.request.Request(url, data=post_data, method='POST')
req.add_header('Content-Type', 'application/json')
try:
    with urllib.request.urlopen(req) as response:
        response_data = response.read().decode('utf-8')
        created_post = json.loads(response_data)
        print("使用 urllib 创建帖子:")
        print(json.dumps(created_post, indent=4, ensure_ascii=False))
except Exception as err:
    print(f"发生错误: {err}")

高级主题

会话对象 (Session)

如果你需要向同一个主机发送多个请求,使用 requests.Session 对象会非常高效,它会保持一个连接池,并自动处理 cookies。

import requests
# 创建一个 Session 对象
with requests.Session() as session:
    # 设置会话级别的 headers (所有该 session 的请求都会带上这些 headers)
    session.headers.update({'User-Agent': 'My-Python-App/1.0'})
    # 第一次请求 (可能设置了 cookie)
    response1 = session.get('https://httpbin.org/cookies/set/test/123')
    print("第一次请求状态码:", response1.status_code)
    # 第二次请求会自动带上第一次设置的 cookie
    response2 = session.get('https://httpbin.org/cookies')
    print("第二次请求获取到的 cookies:", response2.json())

异步请求 (aiohttp)

对于需要高并发请求的场景(例如爬取大量网页),同步的 requests 会成为性能瓶颈,这时可以使用异步库 aiohttp

首先安装:

pip install aiohttp

示例:异步并发获取多个用户数据

import aiohttp
import asyncio
async def fetch_user(session, user_id):
    url = f"https://jsonplaceholder.typicode.com/users/{user_id}"
    try:
        async with session.get(url) as response:
            response.raise_for_status()
            user_data = await response.json() # 使用 await 获取异步解析的 JSON
            print(f"获取用户 {user_id} 成功")
            return user_data
    except aiohttp.ClientError as err:
        print(f"获取用户 {user_id} 失败: {err}")
        return None
async def main():
    # 创建一个异步的 ClientSession
    async with aiohttp.ClientSession() as session:
        # 创建多个任务的列表
        tasks = [fetch_user(session, i) for i in range(1, 6)] # 获取用户 1 到 5
        # 使用 asyncio.gather 并发执行所有任务
        users = await asyncio.gather(*tasks)
        # 过滤掉失败的任务 (返回 None 的)
        successful_users = [user for user in users if user is not None]
        print("\n所有获取到的用户:")
        for user in successful_users:
            print(f"- {user['name']}")
# 运行异步主函数
asyncio.run(main())

总结与最佳实践

特性 requests urllib aiohttp
易用性 非常高 中等
功能 强大,支持会话、Cookie、文件上传等 基础功能 专注于异步和高性能
性能 同步,单线程 同步,单线程 异步,高并发
依赖 需要安装 (pip install requests) Python 标准库,无需安装 需要安装 (pip install aiohttp)
适用场景 绝大多数场景,快速开发 简单脚本,不想安装第三方库 高并发网络请求,如爬虫、微服务通信

最佳实践总结:

  1. 首选 requests:对于绝大多数 API 调用任务,requests 是最佳选择,它简单、强大且文档清晰。
  2. 始终使用 try...except:网络请求是不可靠的,可能会因为各种原因失败(网络问题、服务器错误、API 限流等),务必用 try...except 来捕获 requests.exceptions 下的各种异常。
  3. 检查状态码:使用 response.raise_for_status() 是一个简洁的好方法,它会自动检查状态码并在非 2xx 时抛出异常。
  4. 使用 response.json():API 返回的是 JSON,请使用这个方法来解析,它会自动处理编码问题。
  5. 对于高并发,考虑 aiohttp:当你需要同时处理成百上千个请求时,切换到异步模型可以显著提升性能。
  6. 使用环境变量管理 API Key:不要将 API Key 硬编码在代码中,使用 os.getenv('YOUR_API_KEY') 从环境变量中读取,更安全。
分享:
扫描分享到社交APP
上一篇
下一篇