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

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

wen python案例 223

本文目录导读：

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调用第三方接口的常见场景,您可以根据实际需求选择合适的实现方式。

版权声明：NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播

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

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

相关文章

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

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝