Python中使用cURL进行API调用完整指南

简介

Python作为最受欢迎的编程语言之一，提供了多种方式来执行HTTP请求和与API交互。本文将详细介绍如何在Python中使用cURL命令以及各种HTTP客户端库来进行API调用。

我们将覆盖从基础的subprocess调用cURL，到使用requests、httpx等现代Python库的完整解决方案，帮助你选择最适合项目需求的方法。

Python中执行HTTP请求的方法

主要方法对比

方法	优点	缺点	适用场景
subprocess + cURL	完全兼容cURL命令	性能较低，依赖系统	快速移植现有cURL命令
requests库	简单易用，功能丰富	不支持异步	大部分同步HTTP需求
httpx库	支持异步，现代化API	相对较新	高性能异步应用
urllib (内置)	无需安装额外依赖	API复杂，功能有限	轻量级应用

方法1：使用subprocess调用cURL

基础示例

最直接的方法是使用Python的subprocess模块来执行cURL命令：

import subprocess
import json

def execute_curl(curl_command):
    """执行cURL命令并返回结果"""
    try:
        result = subprocess.run(
            curl_command, 
            shell=True, 
            capture_output=True, 
            text=True, 
            timeout=30
        )
        
        if result.returncode == 0:
            return {
                'success': True,
                'data': result.stdout,
                'error': None
            }
        else:
            return {
                'success': False,
                'data': None,
                'error': result.stderr
            }
    except subprocess.TimeoutExpired:
        return {
            'success': False,
            'data': None,
            'error': 'Request timeout'
        }

# 使用示例
curl_cmd = 'curl -X GET "https://jsonplaceholder.typicode.com/posts/1" -H "Accept: application/json"'
response = execute_curl(curl_cmd)

if response['success']:
    data = json.loads(response['data'])
    print(f"Post title: {data['title']}")
else:
    print(f"Error: {response['error']}")

处理POST请求

import subprocess
import json
import shlex

def post_with_curl(url, data, headers=None):
    """使用cURL执行POST请求"""
    # 构建基础命令
    cmd = ['curl', '-X', 'POST', url]
    
    # 添加数据
    cmd.extend(['-d', json.dumps(data)])
    
    # 添加Content-Type头
    cmd.extend(['-H', 'Content-Type: application/json'])
    
    # 添加其他头
    if headers:
        for key, value in headers.items():
            cmd.extend(['-H', f'{key}: {value}'])
    
    try:
        result = subprocess.run(
            cmd, 
            capture_output=True, 
            text=True, 
            timeout=30
        )
        
        if result.returncode == 0:
            return json.loads(result.stdout)
        else:
            raise Exception(f"cURL error: {result.stderr}")
            
    except subprocess.TimeoutExpired:
        raise Exception("Request timeout")

# 使用示例
post_data = {
    "title": "My New Post",
    "body": "This is the content of my post",
    "userId": 1
}

response = post_with_curl(
    "https://jsonplaceholder.typicode.com/posts",
    post_data,
    {"Authorization": "Bearer your-token"}
)

print(f"Created post ID: {response['id']}")

安全提示：使用subprocess执行shell命令时要特别注意输入验证，避免命令注入攻击。建议使用列表形式的命令而不是字符串。

方法2：使用requests库

安装和基础使用

requests是Python最受欢迎的HTTP库，使用前需要先安装：

pip install requests

GET请求示例

import requests
import json

def get_request_example():
    """使用requests执行GET请求"""
    url = "https://jsonplaceholder.typicode.com/posts"
    
    # 设置请求头
    headers = {
        'Accept': 'application/json',
        'User-Agent': 'Python-App/1.0'
    }
    
    # 设置查询参数
    params = {
        'userId': 1,
        'limit': 10
    }
    
    try:
        response = requests.get(
            url, 
            headers=headers, 
            params=params, 
            timeout=30
        )
        
        # 检查响应状态
        response.raise_for_status()
        
        # 解析JSON响应
        data = response.json()
        
        print(f"Found {len(data)} posts")
        for post in data[:3]:  # 显示前3个
            print(f"- {post['title']}")
            
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")

get_request_example()

POST请求示例

import requests

def post_request_example():
    """使用requests执行POST请求"""
    url = "https://jsonplaceholder.typicode.com/posts"
    
    # 请求数据
    post_data = {
        "title": "My Python Post",
        "body": "Created using Python requests library",
        "userId": 1
    }
    
    # 请求头
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    }
    
    try:
        response = requests.post(
            url,
            json=post_data,  # 自动设置Content-Type和序列化JSON
            headers=headers,
            timeout=30
        )
        
        response.raise_for_status()
        result = response.json()
        
        print(f"Created post with ID: {result['id']}")
        print(f"Response status: {response.status_code}")
        
    except requests.exceptions.RequestException as e:
        print(f"POST request failed: {e}")

post_request_example()

认证和会话管理

import requests
from requests.auth import HTTPBasicAuth

class APIClient:
    """API客户端类，支持认证和会话管理"""
    
    def __init__(self, base_url, auth_token=None):
        self.base_url = base_url
        self.session = requests.Session()
        
        # 设置通用头
        self.session.headers.update({
            'User-Agent': 'Python-API-Client/1.0',
            'Accept': 'application/json'
        })
        
        # 设置认证
        if auth_token:
            self.session.headers['Authorization'] = f'Bearer {auth_token}'
    
    def get(self, endpoint, params=None):
        """GET请求"""
        url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
        try:
            response = self.session.get(url, params=params, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            raise Exception(f"GET request failed: {e}")
    
    def post(self, endpoint, data=None):
        """POST请求"""
        url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
        try:
            response = self.session.post(url, json=data, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            raise Exception(f"POST request failed: {e}")
    
    def close(self):
        """关闭会话"""
        self.session.close()

# 使用示例
client = APIClient("https://jsonplaceholder.typicode.com", "your-api-token")

try:
    # 获取所有文章
    posts = client.get("posts", params={"_limit": 5})
    print(f"Retrieved {len(posts)} posts")
    
    # 创建新文章
    new_post = {
        "title": "Test Post",
        "body": "This is a test post",
        "userId": 1
    }
    
    created_post = client.post("posts", new_post)
    print(f"Created post: {created_post['title']}")
    
finally:
    client.close()

方法3：使用httpx库（异步支持）

安装httpx

pip install httpx

同步使用httpx

import httpx

def httpx_sync_example():
    """使用httpx进行同步HTTP请求"""
    with httpx.Client() as client:
        # GET请求
        response = client.get("https://jsonplaceholder.typicode.com/posts/1")
        post = response.json()
        print(f"Post title: {post['title']}")
        
        # POST请求
        new_post = {
            "title": "HTTPX Post",
            "body": "Created with httpx library",
            "userId": 1
        }
        
        response = client.post(
            "https://jsonplaceholder.typicode.com/posts",
            json=new_post
        )
        
        created = response.json()
        print(f"Created post ID: {created['id']}")

httpx_sync_example()

异步使用httpx

import httpx
import asyncio

async def httpx_async_example():
    """使用httpx进行异步HTTP请求"""
    async with httpx.AsyncClient() as client:
        # 并发执行多个请求
        tasks = []
        for i in range(1, 6):
            task = client.get(f"https://jsonplaceholder.typicode.com/posts/{i}")
            tasks.append(task)
        
        responses = await asyncio.gather(*tasks)
        
        for response in responses:
            post = response.json()
            print(f"Post {post['id']}: {post['title']}")

# 运行异步示例
asyncio.run(httpx_async_example())

高级异步API客户端

import httpx
import asyncio
from typing import List, Dict, Any

class AsyncAPIClient:
    """异步API客户端"""
    
    def __init__(self, base_url: str, auth_token: str = None):
        self.base_url = base_url
        self.headers = {
            'Accept': 'application/json',
            'User-Agent': 'Python-Async-Client/1.0'
        }
        
        if auth_token:
            self.headers['Authorization'] = f'Bearer {auth_token}'
    
    async def get_multiple(self, endpoints: List[str]) -> List[Dict[Any, Any]]:
        """并发获取多个端点的数据"""
        async with httpx.AsyncClient() as client:
            tasks = []
            for endpoint in endpoints:
                url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
                task = client.get(url, headers=self.headers, timeout=30)
                tasks.append(task)
            
            responses = await asyncio.gather(*tasks, return_exceptions=True)
            results = []
            
            for i, response in enumerate(responses):
                if isinstance(response, Exception):
                    results.append({'error': str(response), 'endpoint': endpoints[i]})
                else:
                    try:
                        results.append(response.json())
                    except Exception as e:
                        results.append({'error': f'JSON decode error: {e}', 'endpoint': endpoints[i]})
            
            return results
    
    async def post_batch(self, endpoint: str, data_list: List[Dict]) -> List[Dict]:
        """批量POST请求"""
        async with httpx.AsyncClient() as client:
            url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
            tasks = []
            
            for data in data_list:
                task = client.post(url, json=data, headers=self.headers, timeout=30)
                tasks.append(task)
            
            responses = await asyncio.gather(*tasks, return_exceptions=True)
            results = []
            
            for response in responses:
                if isinstance(response, Exception):
                    results.append({'error': str(response)})
                else:
                    try:
                        results.append(response.json())
                    except Exception as e:
                        results.append({'error': f'JSON decode error: {e}'})
            
            return results

# 使用示例
async def main():
    client = AsyncAPIClient("https://jsonplaceholder.typicode.com")
    
    # 并发获取多个文章
    endpoints = ["posts/1", "posts/2", "posts/3", "posts/4", "posts/5"]
    posts = await client.get_multiple(endpoints)
    
    print("Retrieved posts:")
    for post in posts:
        if 'error' not in post:
            print(f"- {post['title']}")
        else:
            print(f"- Error: {post['error']}")
    
    # 批量创建文章
    new_posts = [
        {"title": f"Async Post {i}", "body": f"Content {i}", "userId": 1}
        for i in range(1, 4)
    ]
    
    created_posts = await client.post_batch("posts", new_posts)
    print(f"\nCreated {len(created_posts)} posts")

# 运行示例
asyncio.run(main())

方法4：使用内置urllib库

基础GET请求

import urllib.request
import urllib.parse
import json

def urllib_get_example():
    """使用urllib执行GET请求"""
    url = "https://jsonplaceholder.typicode.com/posts"
    
    # 添加查询参数
    params = {'userId': 1}
    query_string = urllib.parse.urlencode(params)
    full_url = f"{url}?{query_string}"
    
    # 创建请求对象
    req = urllib.request.Request(
        full_url,
        headers={
            'Accept': 'application/json',
            'User-Agent': 'Python-urllib/1.0'
        }
    )
    
    try:
        with urllib.request.urlopen(req, timeout=30) as response:
            if response.status == 200:
                data = json.loads(response.read().decode('utf-8'))
                print(f"Found {len(data)} posts for user 1")
                return data
            else:
                print(f"Request failed with status: {response.status}")
                return None
                
    except urllib.error.URLError as e:
        print(f"URL Error: {e}")
        return None

urllib_get_example()

POST请求示例

import urllib.request
import urllib.parse
import json

def urllib_post_example():
    """使用urllib执行POST请求"""
    url = "https://jsonplaceholder.typicode.com/posts"
    
    # 准备POST数据
    post_data = {
        "title": "urllib Post",
        "body": "Created using urllib library",
        "userId": 1
    }
    
    # 转换为JSON字节串
    json_data = json.dumps(post_data).encode('utf-8')
    
    # 创建请求对象
    req = urllib.request.Request(
        url,
        data=json_data,
        headers={
            'Content-Type': 'application/json',
            'Accept': 'application/json',
            'User-Agent': 'Python-urllib/1.0'
        },
        method='POST'
    )
    
    try:
        with urllib.request.urlopen(req, timeout=30) as response:
            if response.status in [200, 201]:
                result = json.loads(response.read().decode('utf-8'))
                print(f"Created post with ID: {result['id']}")
                return result
            else:
                print(f"POST failed with status: {response.status}")
                return None
                
    except urllib.error.URLError as e:
        print(f"POST Error: {e}")
        return None

urllib_post_example()

错误处理和重试机制

通用错误处理类

import requests
import time
from functools import wraps

class HTTPClientError(Exception):
    """HTTP客户端错误基类"""
    pass

class RetryableHTTPClient:
    """支持重试的HTTP客户端"""
    
    def __init__(self, max_retries=3, backoff_factor=1):
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor
        self.session = requests.Session()
    
    def retry_on_failure(func):
        """重试装饰器"""
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            last_exception = None
            
            for attempt in range(self.max_retries + 1):
                try:
                    return func(self, *args, **kwargs)
                except requests.exceptions.RequestException as e:
                    last_exception = e
                    if attempt < self.max_retries:
                        wait_time = self.backoff_factor * (2 ** attempt)
                        print(f"Request failed (attempt {attempt + 1}), retrying in {wait_time}s...")
                        time.sleep(wait_time)
                    else:
                        raise HTTPClientError(f"Max retries exceeded: {e}")
            
            raise HTTPClientError(f"Request failed: {last_exception}")
        return wrapper
    
    @retry_on_failure
    def get(self, url, **kwargs):
        """GET请求带重试"""
        response = self.session.get(url, timeout=30, **kwargs)
        response.raise_for_status()
        return response
    
    @retry_on_failure
    def post(self, url, **kwargs):
        """POST请求带重试"""
        response = self.session.post(url, timeout=30, **kwargs)
        response.raise_for_status()
        return response

# 使用示例
client = RetryableHTTPClient(max_retries=3, backoff_factor=1)

try:
    response = client.get("https://jsonplaceholder.typicode.com/posts/1")
    data = response.json()
    print(f"Successfully retrieved: {data['title']}")
except HTTPClientError as e:
    print(f"Failed to retrieve data: {e}")

性能优化技巧

连接池和会话复用

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

class OptimizedHTTPClient:
    """优化的HTTP客户端"""
    
    def __init__(self):
        self.session = requests.Session()
        
        # 配置重试策略
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
        )
        
        # 配置连接适配器
        adapter = HTTPAdapter(
            max_retries=retry_strategy,
            pool_connections=20,  # 连接池大小
            pool_maxsize=20       # 最大连接数
        )
        
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
        
        # 设置通用头
        self.session.headers.update({
            'User-Agent': 'Optimized-Python-Client/1.0',
            'Connection': 'keep-alive'
        })
    
    def batch_get(self, urls):
        """批量GET请求"""
        results = []
        for url in urls:
            try:
                response = self.session.get(url, timeout=30)
                response.raise_for_status()
                results.append({
                    'url': url,
                    'status': response.status_code,
                    'data': response.json()
                })
            except Exception as e:
                results.append({
                    'url': url,
                    'error': str(e)
                })
        return results
    
    def close(self):
        """关闭会话"""
        self.session.close()

# 使用示例
client = OptimizedHTTPClient()

urls = [
    "https://jsonplaceholder.typicode.com/posts/1",
    "https://jsonplaceholder.typicode.com/posts/2", 
    "https://jsonplaceholder.typicode.com/posts/3"
]

results = client.batch_get(urls)
for result in results:
    if 'error' not in result:
        print(f"Successfully retrieved post: {result['data']['title']}")
    else:
        print(f"Failed to retrieve {result['url']}: {result['error']}")

client.close()

最佳实践

1. 选择合适的方法

简单脚本：使用requests库
高性能应用：使用httpx或aiohttp
无依赖需求：使用urllib
快速移植cURL：使用subprocess

2. 安全考虑

安全建议：

始终验证SSL证书（避免使用verify=False）
设置合理的超时时间
不要在代码中硬编码敏感信息
使用环境变量存储API密钥
验证输入参数，防止注入攻击

3. 错误处理策略

import os
import requests
from requests.exceptions import RequestException

def secure_api_call(url, method='GET', data=None):
    """安全的API调用示例"""
    
    # 从环境变量获取API密钥
    api_key = os.getenv('API_KEY')
    if not api_key:
        raise ValueError("API_KEY environment variable not set")
    
    headers = {
        'Authorization': f'Bearer {api_key}',
        'Content-Type': 'application/json',
        'User-Agent': 'Secure-Python-Client/1.0'
    }
    
    try:
        if method.upper() == 'GET':
            response = requests.get(url, headers=headers, timeout=30, verify=True)
        elif method.upper() == 'POST':
            response = requests.post(url, json=data, headers=headers, timeout=30, verify=True)
        else:
            raise ValueError(f"Unsupported method: {method}")
        
        # 检查HTTP状态码
        response.raise_for_status()
        
        # 验证响应内容类型
        content_type = response.headers.get('content-type', '')
        if 'application/json' not in content_type:
            raise ValueError(f"Unexpected content type: {content_type}")
        
        return response.json()
        
    except RequestException as e:
        # 记录错误但不暴露敏感信息
        print(f"API request failed: {type(e).__name__}")
        raise
    except ValueError as e:
        print(f"Value error: {e}")
        raise
    except Exception as e:
        print(f"Unexpected error: {type(e).__name__}")
        raise

# 使用示例（需要设置环境变量）
# export API_KEY="your-secret-key"
try:
    result = secure_api_call("https://api.example.com/data")
    print("API call successful")
except Exception as e:
    print(f"API call failed: {e}")

总结

Python提供了多种强大的HTTP请求方法，每种都有其适用场景：

requests - 最受欢迎，功能丰富，适合大多数同步应用
httpx - 现代化设计，支持异步，适合高性能应用
urllib - 内置库，适合轻量级应用
subprocess + cURL - 适合快速移植现有cURL命令

选择合适的方法时，要考虑项目需求、性能要求、依赖管理和团队熟悉度。无论选择哪种方法，都要注意安全性、错误处理和性能优化。