IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Python脚本如何调用RESTful API

wen 实用脚本 1

本文目录导读：

Python脚本如何调用RESTful API

  1. 使用 requests 库（最常用）
  2. 完整的API调用类示例
  3. 处理认证
  4. 错误处理和重试
  5. 异步调用（使用 aiohttp）
  6. 实用工具函数
  7. 注意事项

我来详细说明Python调用RESTful API的几种方法：

使用 requests 库（最常用）

安装

pip install requests

基本用法示例

import requests
import json
# GET 请求
def get_example():
    url = "https://api.example.com/users"
    headers = {
        "Authorization": "Bearer YOUR_TOKEN",
        "Content-Type": "application/json"
    }
    params = {
        "page": 1,
        "limit": 10
    }
    response = requests.get(url, headers=headers, params=params)
    if response.status_code == 200:
        data = response.json()
        return data
    else:
        print(f"Error: {response.status_code}")
        return None
# POST 请求
def post_example():
    url = "https://api.example.com/users"
    headers = {
        "Authorization": "Bearer YOUR_TOKEN",
        "Content-Type": "application/json"
    }
    data = {
        "name": "John Doe",
        "email": "john@example.com",
        "age": 30
    }
    response = requests.post(url, headers=headers, json=data)
    return response.json()
# PUT 请求
def put_example():
    url = "https://api.example.com/users/1"
    data = {
        "name": "Jane Doe",
        "email": "jane@example.com"
    }
    response = requests.put(url, json=data)
    return response.json()
# DELETE 请求
def delete_example():
    url = "https://api.example.com/users/1"
    response = requests.delete(url)
    if response.status_code == 204:
        print("Successfully deleted")
    else:
        print(f"Error: {response.status_code}")

完整的API调用类示例

import requests
import json
from typing import Dict, Any, Optional
class APIClient:
    def __init__(self, base_url: str, api_key: str = None):
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "Content-Type": "application/json",
            "Accept": "application/json"
        })
        if api_key:
            self.session.headers.update({
                "Authorization": f"Bearer {api_key}"
            })
    def get(self, endpoint: str, params: Dict = None) -> Dict:
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = self.session.get(url, params=params)
        self._handle_response(response)
        return response.json()
    def post(self, endpoint: str, data: Dict = None) -> Dict:
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = self.session.post(url, json=data)
        self._handle_response(response)
        return response.json()
    def put(self, endpoint: str, data: Dict = None) -> Dict:
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = self.session.put(url, json=data)
        self._handle_response(response)
        return response.json()
    def delete(self, endpoint: str) -> bool:
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = self.session.delete(url)
        self._handle_response(response)
        return response.status_code == 204
    def _handle_response(self, response):
        if response.status_code >= 400:
            raise Exception(f"API Error: {response.status_code} - {response.text}")
    def close(self):
        self.session.close()
# 使用示例
def api_client_example():
    client = APIClient("https://api.example.com", "your_api_key_here")
    try:
        # 获取用户列表
        users = client.get("users", {"page": 1})
        print(users)
        # 创建新用户
        new_user = client.post("users", {
            "name": "Alice",
            "email": "alice@example.com"
        })
        print(new_user)
        # 更新用户
        updated_user = client.put("users/1", {"name": "Bob"})
        print(updated_user)
        # 删除用户
        result = client.delete("users/1")
        print(f"Deleted: {result}")
    finally:
        client.close()

处理认证

import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
# 基本认证
def basic_auth_example():
    url = "https://api.example.com/login"
    auth = HTTPBasicAuth("username", "password")
    response = requests.get(url, auth=auth)
    return response.json()
# Token 认证
def token_auth_example():
    url = "https://api.example.com/data"
    headers = {
        "Authorization": "Token YOUR_TOKEN_HERE",
        "Content-Type": "application/json"
    }
    response = requests.get(url, headers=headers)
    return response.json()
# OAuth2 认证
def oauth2_example():
    # 先获取 access token
    token_url = "https://api.example.com/oauth/token"
    client_id = "your_client_id"
    client_secret = "your_client_secret"
    token_response = requests.post(token_url, data={
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret
    })
    if token_response.status_code == 200:
        access_token = token_response.json()["access_token"]
        # 使用 token 调用 API
        url = "https://api.example.com/protected-resource"
        headers = {"Authorization": f"Bearer {access_token}"}
        response = requests.get(url, headers=headers)
        return response.json()

错误处理和重试

import requests
from time import sleep
from functools import wraps
def retry(max_retries=3, delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except (requests.ConnectionError, requests.Timeout) as e:
                    if attempt == max_retries - 1:
                        raise e
                    print(f"Retry attempt {attempt + 1} after error: {e}")
                    sleep(delay * (attempt + 1))
            return None
        return wrapper
    return decorator
@retry(max_retries=3, delay=2)
def fetch_data_with_retry():
    url = "https://api.example.com/unstable-endpoint"
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    return response.json()
# 使用示例
try:
    data = fetch_data_with_retry()
    print(f"Data: {data}")
except Exception as e:
    print(f"Failed after retries: {e}")

异步调用（使用 aiohttp）

import aiohttp
import asyncio
async def async_api_call():
    async with aiohttp.ClientSession() as session:
        # GET 请求
        async with session.get('https://api.example.com/users') as response:
            data = await response.json()
            print(f"Users: {data}")
        # POST 请求
        async with session.post(
            'https://api.example.com/users',
            json={'name': 'Test', 'email': 'test@example.com'}
        ) as response:
            data = await response.json()
            print(f"Created: {data}")
# 运行异步函数
asyncio.run(async_api_call())

实用工具函数

import requests
from urllib.parse import urlencode
def build_url(base_url, endpoint, params=None):
    url = f"{base_url.rstrip('/')}/{endpoint.lstrip('/')}"
    if params:
        url += f"?{urlencode(params)}"
    return url
def api_response_check(response):
    """检查API响应并返回格式化的结果"""
    try:
        data = response.json()
        return {
            'success': response.ok,
            'status_code': response.status_code,
            'data': data,
            'headers': dict(response.headers)
        }
    except ValueError:
        return {
            'success': response.ok,
            'status_code': response.status_code,
            'data': response.text,
            'headers': dict(response.headers)
        }
# 使用示例
response = requests.get('https://api.example.com/users')
result = api_response_check(response)
print(f"API Response: {result}")

注意事项

  1. 错误处理：始终使用 try-except 捕获异常
  2. 超时设置：设置合理的超时时间 timeout=10
  3. SSL 验证：生产环境保持 SSL 验证，开发时可临时禁用
  4. 会话管理：多次请求使用 Session 对象提高性能
  5. 资源清理：确保正确关闭连接
# 推荐的生产环境配置
def production_ready_call():
    session = requests.Session()
    session.headers.update({
        'User-Agent': 'MyApp/1.0',
        'Accept': 'application/json'
    })
    try:
        response = session.get(
            'https://api.example.com/data',
            timeout=(3.05, 27),  # (connect timeout, read timeout)
            verify=True  # SSL verification
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        return None
    finally:
        session.close()

选择哪种方法取决于你的具体需求,requests 库是最常用且功能完整的选项。

上一篇监控硬盘健康状态的脚本如何写

下一篇Shell脚本如何实现管道重定向

相关文章

抱歉，评论功能暂时关闭!