当前位置:知识文库 ❯ 图文

Python RESTful API调用教程 - JSON交互与认证

一、RESTful API 概述

REST（Representational State Transfer，表述性状态转移）是一种软件架构风格，由 Roy Fielding 在 2000 年的博士论文中首次提出。RESTful API 是指遵循 REST 原则设计的 Web API，它使用 HTTP 协议的标准方法（GET、POST、PUT、DELETE 等）对资源进行操作。

如今，绝大多数主流服务都提供了 RESTful API，包括 GitHub API、Twitter API、Stripe API 等。掌握 RESTful API 调用能力，是 Python 开发者必备的技能之一。

二、REST 架构核心原则

资源与 HTTP 方法映射

REST 将一切视为资源，每个资源通过唯一的 URI 标识。对资源的操作通过 HTTP 方法来表达：

HTTP 方法	操作含义	对应 CRUD	示例 URI
`GET`	获取资源	Read	GET /api/users
`POST`	创建资源	Create	POST /api/users
`PUT`	完整更新资源	Update	PUT /api/users/1
`PATCH`	部分更新资源	Update	PATCH /api/users/1
`DELETE`	删除资源	Delete	DELETE /api/users/1

REST 的核心约束

无状态性：每次请求包含所有必要信息，服务端不保存客户端上下文
统一接口：通过标准 HTTP 方法操作资源，接口行为可预测
资源标识：每个资源通过唯一 URI 定位
表述性传输：客户端通过 JSON/XML 等格式与服务端交换资源状态

三、JSON 数据交互

JSON（JavaScript Object Notation）是 RESTful API 最常用的数据格式。Python 内置的 json 模块和 requests 库的自动序列化功能让 JSON 处理变得非常简单。

发送 JSON 请求体

代码示例

import requests
import json

# 使用 json 参数自动序列化（推荐）
user_data = {
    "name": "张三",
    "email": "zhangsan@example.com",
    "roles": ["admin", "editor"],
    "active": True
}

response = requests.post(
    'https://httpbin.org/post',
    json=user_data,
    headers={'Content-Type': 'application/json'}
)

# 解析 JSON 响应
result = response.json()
print(f'发送的数据：{result["json"]}')

# 手动序列化（适用于需要自定义序列化逻辑的场景）
response2 = requests.post(
    'https://httpbin.org/post',
    data=json.dumps(user_data, ensure_ascii=False),
    headers={'Content-Type': 'application/json'}
)

处理复杂 JSON 响应

代码示例

import requests

response = requests.get('https://jsonplaceholder.typicode.com/users')
users = response.json()

# 遍历嵌套数据结构
for user in users:
    name = user['name']
    email = user['email']
    company = user['company']['name']
    print(f'{name} ({email}) - 公司：{company}')

# 使用列表推导式提取特定字段
user_emails = [u['email'] for u in users if u['address']['city'] == 'South Elvis']
print(f'符合条件的邮箱：{user_emails}')

提示：处理不确定的 JSON 结构时，推荐使用 .get('key', 'default') 方法或 try-except KeyError 来避免 KeyError 异常。

四、认证方式详解

API Key 认证

API Key 是最简单的认证方式，通常通过 URL 参数或请求头传递：

代码示例

import requests

API_KEY = 'your-api-key-here'

# 方式一：通过请求头传递（推荐）
headers = {
    'X-API-Key': API_KEY,
    'Authorization': f'ApiKey {API_KEY}'
}
response = requests.get('https://api.example.com/data', headers=headers)

# 方式二：通过 URL 参数传递
params = {'api_key': API_KEY}
response = requests.get('https://api.example.com/data', params=params)

Bearer Token 认证（OAuth 2.0）

代码示例

import requests

# 使用 Bearer Token 认证
token = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...'
headers = {
    'Authorization': f'Bearer {token}',
    'Accept': 'application/json'
}

response = requests.get('https://api.example.com/protected-resource', headers=headers)

if response.status_code == 200:
    data = response.json()
    print(f'获取成功：{data}')
elif response.status_code == 401:
    print('Token 无效或已过期，需要重新获取')

基本认证（Basic Auth）

代码示例

import requests
from requests.auth import HTTPBasicAuth

# 方式一：使用 auth 参数
response = requests.get(
    'https://httpbin.org/basic-auth/user/passwd',
    auth=('user', 'passwd')
)

# 方式二：使用 HTTPBasicAuth 对象
response = requests.get(
    'https://httpbin.org/basic-auth/user/passwd',
    auth=HTTPBasicAuth('user', 'passwd')
)

print(f'认证结果：{response.status_code}')

五、CRUD 操作实战

以下使用 JSONPlaceholder 免费测试 API 演示完整的 CRUD 操作：

代码示例

import requests

BASE_URL = 'https://jsonplaceholder.typicode.com'

# ===== 创建（Create） =====
new_post = {
    'title': '我的第一篇博客',
    'body': '这是使用 RESTful API 创建的文章内容。',
    'userId': 1
}
create_resp = requests.post(f'{BASE_URL}/posts', json=new_post)
print(f'创建响应：{create_resp.status_code}')
created_post = create_resp.json()
print(f'创建的文章ID：{created_post["id"]}')

# ===== 读取（Read） =====
# 获取单条记录
get_resp = requests.get(f'{BASE_URL}/posts/1')
post = get_resp.json()
print(f'文章标题：{post["title"]}')

# 获取列表
list_resp = requests.get(f'{BASE_URL}/posts?userId=1')
posts = list_resp.json()
print(f'用户1的文章数量：{len(posts)}')

# ===== 更新（Update） =====
update_data = {
    'id': 1,
    'title': '更新后的标题',
    'body': '更新后的内容',
    'userId': 1
}
put_resp = requests.put(f'{BASE_URL}/posts/1', json=update_data)
print(f'更新响应：{put_resp.status_code}')

# 部分更新（PATCH）
patch_resp = requests.patch(
    f'{BASE_URL}/posts/1',
    json={'title': '只更新标题'}
)
print(f'部分更新响应：{patch_resp.status_code}')

# ===== 删除（Delete） =====
delete_resp = requests.delete(f'{BASE_URL}/posts/1')
print(f'删除响应：{delete_resp.status_code}')

六、分页与过滤参数

当 API 返回大量数据时，通常使用分页机制。常见的分页方式有偏移量分页和游标分页。

偏移量分页（Offset-based）

代码示例

import requests

def fetch_all_pages(base_url, params=None, page_size=100):
    """自动翻页获取所有数据"""
    all_items = []
    page = 1

    while True:
        request_params = {
            **(params or {}),
            'page': page,
            'per_page': page_size
        }

        response = requests.get(base_url, params=request_params)
        response.raise_for_status()

        items = response.json()
        if not items:
            break  # 没有更多数据，退出循环

        all_items.extend(items)
        print(f'已获取第 {page} 页，共 {len(items)} 条')

        if len(items) < page_size:
            break  # 最后一页

        page += 1

    return all_items

# 使用示例：获取 GitHub 上 Python 相关的所有仓库
all_repos = fetch_all_pages(
    'https://api.github.com/search/repositories',
    params={'q': 'language:python', 'sort': 'stars'}
)
print(f'总共获取 {len(all_repos)} 条记录')

过滤与排序参数

代码示例

import requests

# 常见查询参数
params = {
    'status': 'active',        # 状态过滤
    'category': 'technology',  # 分类过滤
    'sort_by': 'created_at',   # 排序字段
    'sort_order': 'desc',      # 排序方向
    'fields': 'id,name,email', # 字段选择
    'q': 'Python教程'          # 关键词搜索
}

response = requests.get('https://api.example.com/articles', params=params)
articles = response.json()

七、错误处理与重试机制

HTTP 状态码处理

状态码	含义	常见场景
200 OK	请求成功	正常返回数据
201 Created	资源创建成功	POST 请求成功创建资源
400 Bad Request	请求参数错误	参数格式不正确
401 Unauthorized	未认证	Token 无效或缺失
403 Forbidden	无权限	权限不足
404 Not Found	资源不存在	URL 错误或资源已删除
429 Too Many Requests	请求过于频繁	触发速率限制
500 Internal Server Error	服务器内部错误	服务端 bug 或过载

自动重试机制

代码示例

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import time

def create_retry_session(
    total_retries=3,
    backoff_factor=0.5,
    status_forcelist=(429, 500, 502, 503, 504)
):
    """创建带自动重试的 Session"""
    session = requests.Session()

    # 配置重试策略
    retry_strategy = Retry(
        total=total_retries,
        backoff_factor=backoff_factor,  # 重试间隔：0.5s, 1s, 2s...
        status_forcelist=status_forcelist,
        allowed_methods=["GET", "POST", "PUT", "DELETE"]
    )

    # 挂载适配器
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount('http://', adapter)
    session.mount('https://', adapter)

    return session

# 使用带重试的 Session
session = create_retry_session(total_retries=3)
response = session.get('https://api.example.com/data', timeout=10)
print(f'响应状态码：{response.status_code}')

提示：429 状态码通常伴随 Retry-After 响应头，指示客户端等待的秒数。在实际项目中应优先读取该值进行等待。

八、实战：调用 GitHub API

以下是一个完整的实战案例，使用 GitHub API 搜索热门 Python 仓库并提取关键信息：

代码示例

import requests
from datetime import datetime

def search_github_repos(query, sort='stars', order='desc', per_page=5):
    """搜索 GitHub 仓库"""
    url = 'https://api.github.com/search/repositories'

    headers = {
        'Accept': 'application/vnd.github.v3+json',
        'User-Agent': 'Python-API-Client'
    }

    params = {
        'q': query,
        'sort': sort,
        'order': order,
        'per_page': per_page
    }

    try:
        response = requests.get(url, headers=headers, params=params, timeout=10)
        response.raise_for_status()

        data = response.json()
        total_count = data['total_count']
        repos = data['items']

        print(f'搜索结果：共 {total_count} 个仓库，显示前 {len(repos)} 个\n')
        print('=' * 60)

        for i, repo in enumerate(repos, 1):
            print(f'{i}. {repo["full_name"]}')
            print(f'   描述：{repo["description"] or "无描述"}')
            print(f'   Stars：{repo["stargazers_count"]:,}')
            print(f'   Forks：{repo["forks_count"]:,}')
            print(f'   语言：{repo["language"] or "未知"}')
            print(f'   更新时间：{repo["updated_at"][:10]}')
            print(f'   链接：{repo["html_url"]}')
            print('-' * 60)

        return repos

    except requests.exceptions.HTTPError as e:
        if response.status_code == 403:
            print('API 速率限制已达，请稍后再试')
        else:
            print(f'HTTP 错误：{e}')
    except requests.exceptions.RequestException as e:
        print(f'请求失败：{e}')

# 运行示例
if __name__ == '__main__':
    search_github_repos('language:python stars:>10000')

九、小结与练习题

核心要点回顾

REST 架构：资源 + HTTP 方法 = 统一接口
JSON 交互：使用 json= 参数自动序列化
认证方式：API Key、Bearer Token、Basic Auth 各有适用场景
CRUD 操作：GET 读、POST 创、PUT/PATCH 改、DELETE 删
重试机制：使用 HTTPAdapter 和 Retry 策略处理临时故障

练习1

使用 JSONPlaceholder API，编写一个类 TodoManager，实现创建、查询、更新、删除 TODO 项目的完整功能。

练习2

编写一个通用 API 客户端类 APIClient，支持配置 base_url、认证方式（Bearer Token）、自动重试、超时设置，并提供 get/post/put/delete 方法封装。

常见问题

REST 和 GraphQL 有什么区别？

REST 基于资源，每个端点对应一个资源，客户端可能需要多次请求获取不同资源。GraphQL 使用单一端点，客户端可以精确指定需要的字段，减少过度获取和多次请求。REST 更简单、缓存友好；GraphQL 更灵活但学习成本更高。

如何安全地存储 API 密钥？

切勿将 API 密钥硬编码在代码中。推荐使用环境变量（os.environ 或 python-dotenv）、密钥管理服务（如 AWS Secrets Manager），或加密配置文件。在 .gitignore 中排除包含密钥的文件，防止泄露到版本控制系统。

PUT 和 PATCH 有什么区别？

PUT 是完整替换资源，需要提供资源的所有字段；PATCH 是部分更新，只需提供要修改的字段。如果只需要修改某个字段，使用 PATCH 更高效且更安全。

API 返回 429 状态码如何处理？

429 表示请求频率超过了 API 的速率限制。应该先读取 Retry-After 响应头获取等待时间，然后休眠相应时间后重试。也可以使用指数退避策略（exponential backoff），每次重试的等待时间成倍增长。

标签： Python RESTful API JSON API认证 CRUD 网络编程

本文涉及AI创作

内容由AI创作，请仔细甄别

list快速访问

poll相关推荐

Django静态文件管理教程：STATIC配置、collectstatic与Nginx部署

Django Migration迁移详解 - 数据库版本控制完整指南

Django ORM查询详解 - 数据库操作与优化完整指南

Django中间件详解 - 请求响应钩子系统完整指南

快速访问

相关推荐