如何在Python3.9中执行指定Curl命令并提取存储access_token

我正在使用Python3.9，现有一段curl命令：

curl -x proxy1.req:8080 --location --request POST 'https://auth.trst.com/2.0' \
--header 'Content-Type: application/x-www-form-urlencoded' \
--data-urlencode 'ct_id=33d3e1-4988-9a08-e7773634ba67' \
--data-urlencode 'ct_sec=56GgtoQA8.WIH9nFM3Eh3sT~cwH' \
--data-urlencode 'scope=https://auth.trst.com/settlement/.default' \
--data-urlencode 'grant_type=client_credentials'

该命令返回的响应包含access_token字段，我需要在Python中实现该请求，并将access_token存入变量用于后续逻辑。

方法1：使用requests库（推荐） #

这是最符合Python开发习惯的方式，无需依赖系统curl，直接发送HTTP请求。

先安装requests库：

pip install requests

编写代码：

import requests

# 配置请求参数
url = 'https://auth.trst.com/2.0'
proxy_config = {
    'http': 'http://proxy1.req:8080',
    'https': 'http://proxy1.req:8080'
}
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
form_data = {
    'ct_id': '33d3e1-4988-9a08-e7773634ba67',
    'ct_sec': '56GgtoQA8.WIH9nFM3Eh3sT~cwH',
    'scope': 'https://auth.trst.com/settlement/.default',
    'grant_type': 'client_credentials'
}

# 发送请求并处理响应
try:
    response = requests.post(url, headers=headers, data=form_data, proxies=proxy_config)
    response.raise_for_status()  # 抛出请求异常
    response_json = response.json()
    access_token = response_json.get('access_token')
    
    if access_token:
        print(f"获取到access_token: {access_token}")
        # 后续业务逻辑直接使用access_token变量即可
    else:
        print("响应中未找到access_token字段")
except requests.exceptions.RequestException as e:
    print(f"请求失败: {str(e)}")

方法2：调用系统curl命令（不推荐） #

如果必须依赖curl命令执行，可通过subprocess模块调用并解析输出：

import subprocess
import json

# 构造curl命令列表
curl_command = [
    'curl', '-x', 'proxy1.req:8080', '--location', '--request', 'POST',
    'https://auth.trst.com/2.0',
    '--header', 'Content-Type: application/x-www-form-urlencoded',
    '--data-urlencode', 'ct_id=33d3e1-4988-9a08-e7773634ba67',
    '--data-urlencode', 'ct_sec=56GgtoQA8.WIH9nFM3Eh3sT~cwH',
    '--data-urlencode', 'scope=https://auth.trst.com/settlement/.default',
    '--data-urlencode', 'grant_type=client_credentials'
]

# 执行命令并处理结果
result = subprocess.run(curl_command, capture_output=True, text=True)
if result.returncode != 0:
    print(f"curl执行失败: {result.stderr}")
else:
    # 跳过curl的进度信息，提取JSON部分
    output_content = result.stdout
    json_start_idx = output_content.find('{')
    if json_start_idx != -1:
        json_str = output_content[json_start_idx:]
        try:
            response_json = json.loads(json_str)
            access_token = response_json.get('access_token')
            if access_token:
                print(f"获取到access_token: {access_token}")
            else:
                print("响应中未找到access_token字段")
        except json.JSONDecodeError as e:
            print(f"JSON解析失败: {str(e)}")
    else:
        print("响应中未找到有效JSON数据")

注意事项 #

• 优先选择方法1，避免依赖系统curl环境，代码更易维护和调试。
• 敏感信息如ct_sec不要硬编码，建议通过环境变量或加密配置文件读取，提升安全性。

内容的提问来源于stack exchange，提问作者Marcus