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

Python接口请求案例如何调用第三方接口

wen python案例 27

本文目录导读：

Python接口请求案例如何调用第三方接口

  1. 基础HTTP请求库
  2. 完整案例：调用天气API
  3. 通用API调用工具类
  4. 带重试机制的API调用
  5. 异步API调用（aiohttp）
  6. 最佳实践建议

我来为您详细介绍Python调用第三方接口的几种常见方法和完整案例。

基础HTTP请求库

使用requests库（最常用）

import requests
import json
# 基础GET请求
response = requests.get('https://api.example.com/users')
print(response.status_code)
print(response.json())
# GET请求带参数
params = {
    'page': 1,
    'limit': 10,
    'search': 'python'
}
response = requests.get('https://api.example.com/users', params=params)
# POST请求
data = {
    'username': 'testuser',
    'email': 'test@example.com'
}
headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer your_token_here'
}
response = requests.post('https://api.example.com/users', 
                        json=data, 
                        headers=headers)
# 带认证的请求
from requests.auth import HTTPBasicAuth
response = requests.get('https://api.example.com/secure',
                        auth=HTTPBasicAuth('username', 'password'))

使用urllib（标准库）

import urllib.request
import urllib.parse
import json
# POST请求示例
url = 'https://api.example.com/data'
data = {'key': 'value'}
headers = {'Content-Type': 'application/json'}
# 将数据编码为JSON
data_json = json.dumps(data).encode('utf-8')
req = urllib.request.Request(url, data=data_json, headers=headers)
try:
    with urllib.request.urlopen(req) as response:
        result = json.loads(response.read().decode('utf-8'))
        print(result)
except urllib.error.URLError as e:
    print(f"请求失败: {e}")

完整案例：调用天气API

import requests
import json
from datetime import datetime
class WeatherAPI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.openweathermap.org/data/2.5"
    def get_weather_by_city(self, city_name):
        """根据城市名称获取天气"""
        endpoint = f"{self.base_url}/weather"
        params = {
            'q': city_name,
            'appid': self.api_key,
            'units': 'metric',  # 公制单位
            'lang': 'zh_cn'     # 中文
        }
        try:
            response = requests.get(endpoint, params=params, timeout=10)
            response.raise_for_status()  # 检查HTTP错误
            data = response.json()
            return self._format_weather_data(data)
        except requests.exceptions.RequestException as e:
            return {"error": f"请求失败: {str(e)}"}
        except json.JSONDecodeError:
            return {"error": "数据解析失败"}
    def _format_weather_data(self, data):
        """格式化天气数据"""
        return {
            'city': data['name'],
            'temperature': data['main']['temp'],
            'feels_like': data['main']['feels_like'],
            'humidity': data['main']['humidity'],
            'description': data['weather'][0]['description'],
            'wind_speed': data['wind']['speed'],
            'timestamp': datetime.fromtimestamp(data['dt']).strftime('%Y-%m-%d %H:%M:%S')
        }
# 使用示例
def main():
    # 需要替换为真实的API密钥
    api_key = "your_openweather_api_key"
    weather = WeatherAPI(api_key)
    # 获取天气
    result = weather.get_weather_by_city("Beijing")
    if 'error' in result:
        print(f"错误: {result['error']}")
    else:
        print(f"城市: {result['city']}")
        print(f"温度: {result['temperature']}°C")
        print(f"体感温度: {result['feels_like']}°C")
        print(f"湿度: {result['humidity']}%")
        print(f"天气: {result['description']}")
        print(f"风速: {result['wind_speed']}m/s")
        print(f"更新时间: {result['timestamp']}")
if __name__ == "__main__":
    main()

通用API调用工具类

import requests
import json
import time
from typing import Dict, Any, Optional
class APIClient:
    def __init__(self, base_url: str, api_key: Optional[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[str, Any]:
        """GET请求"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        try:
            response = self.session.get(url, params=params, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.Timeout:
            return {"error": "请求超时"}
        except requests.exceptions.HTTPError as e:
            return {"error": f"HTTP错误: {e.response.status_code}"}
        except Exception as e:
            return {"error": f"请求失败: {str(e)}"}
    def post(self, endpoint: str, data: Dict = None) -> Dict[str, Any]:
        """POST请求"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        try:
            response = self.session.post(url, json=data, timeout=30)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            return {"error": f"请求失败: {str(e)}"}
    def put(self, endpoint: str, data: Dict = None) -> Dict[str, Any]:
        """PUT请求"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        try:
            response = self.session.put(url, json=data, timeout=30)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            return {"error": f"请求失败: {str(e)}"}
    def delete(self, endpoint: str) -> Dict[str, Any]:
        """DELETE请求"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        try:
            response = self.session.delete(url, timeout=30)
            response.raise_for_status()
            return response.json() if response.content else {"success": True}
        except Exception as e:
            return {"error": f"请求失败: {str(e)}"}
    def close(self):
        """关闭会话"""
        self.session.close()
# 使用示例
def api_example():
    # 创建客户端
    client = APIClient(
        base_url="https://jsonplaceholder.typicode.com",
        api_key=None  # 不需要认证
    )
    try:
        # GET请求
        users = client.get("/users")
        print(f"获取到 {len(users)} 个用户")
        # POST请求
        new_user = client.post("/users", {
            "name": "John Doe",
            "email": "john@example.com"
        })
        print(f"创建用户: {new_user.get('name')}")
        # PUT请求
        updated_user = client.put("/users/1", {
            "name": "Jane Doe",
            "email": "jane@example.com"
        })
        print(f"更新用户: {updated_user.get('name')}")
    except Exception as e:
        print(f"API调用失败: {e}")
    finally:
        client.close()
if __name__ == "__main__":
    api_example()

带重试机制的API调用

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 i in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except requests.exceptions.RequestException as e:
                    if i == max_retries - 1:
                        raise
                    print(f"请求失败，正在重试 ({i+1}/{max_retries})...")
                    sleep(delay * (i + 1))  # 指数退避
            return None
        return wrapper
    return decorator
class ReliableAPIClient:
    @retry(max_retries=3, delay=2)
    def make_request(self, url, method='GET', **kwargs):
        """带重试的请求"""
        try:
            if method.upper() == 'GET':
                response = requests.get(url, **kwargs)
            elif method.upper() == 'POST':
                response = requests.post(url, **kwargs)
            else:
                raise ValueError(f"不支持的请求方法: {method}")
            response.raise_for_status()
            return response.json()
        except requests.exceptions.Timeout:
            raise
        except requests.exceptions.ConnectionError:
            raise
        except requests.exceptions.HTTPError:
            raise
# 使用示例
def reliable_api_example():
    client = ReliableAPIClient()
    try:
        # 调用API
        url = "https://api.example.com/data"
        result = client.make_request(url, method='GET')
        print(result)
    except requests.exceptions.RequestException as e:
        print(f"最终请求失败: {e}")
if __name__ == "__main__":
    reliable_api_example()

异步API调用（aiohttp）

import aiohttp
import asyncio
import json
async def async_api_call(session, url, method='GET', data=None):
    """异步API调用"""
    try:
        async with session.request(method, url, json=data) as response:
            if response.status == 200:
                return await response.json()
            else:
                return {
                    "error": f"HTTP {response.status}",
                    "message": await response.text()
                }
    except aiohttp.ClientError as e:
        return {"error": str(e)}
async def fetch_multiple_apis():
    """并发调用多个API"""
    async with aiohttp.ClientSession() as session:
        tasks = [
            async_api_call(session, 'https://api1.example.com/data'),
            async_api_call(session, 'https://api2.example.com/info'),
            async_api_call(session, 'https://api3.example.com/status')
        ]
        results = await asyncio.gather(*tasks)
        return results
# 使用示例
def async_example():
    results = asyncio.run(fetch_multiple_apis())
    for i, result in enumerate(results, 1):
        print(f"API {i} 结果: {result}")
if __name__ == "__main__":
    async_example()

最佳实践建议

  1. 错误处理：始终处理网络错误、超时和HTTP错误码
  2. 超时设置：设置合理的超时时间，避免长时间阻塞
  3. 重试机制：对临时性错误实施指数退避重试策略
  4. 会话复用：使用Session对象复用TCP连接
  5. 环境变量：敏感信息（API密钥）使用环境变量存储
  6. 限速处理：遵守API的速率限制，必要时实现请求队列
  7. 日志记录：记录API调用日志，便于问题排查
  8. 单元测试：为API调用编写测试用例，使用mock模拟外部请求

这些案例涵盖了Python调用第三方接口的常见场景,您可以根据实际需求选择合适的实现方式。

上一篇Python接口返回案例如何解析返回数据

下一篇Python接口参数案例如何拼接请求参数

相关文章

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