使用Python subprocess执行Elasticsearch查询时触发Illegal Argument Exception

原可正常执行的curl命令 #

curl -k XGET -H "Content-Type: application/json" -H "Authorization: ApiKey Q2txZ2FvSUJ6Nlcwa3pjbnh0NUM6enBidjdNLTdRQVNhc0tuTEk5UEZmZw==" 'https://internal-a02c1b85dade940ef871b0b6d1e65191-1270244841.us-west-2.elb.amazonaws.com:9200/_search?index=.fleet-enrollment-api-keys-7&size=20&pretty' -d '{"query": {"match": {"_id": "c8a537a9-7f98-489b-8f8b-de2450c67ce6"}},"fields" : ["_id", "_index"]}'

出错的Python代码 #

import json
from builtins import int
from pydantic import BaseModel, Field
from typing import List, Dict
from subprocess import PIPE, run

def elasticsearch_search_query(host: str, 
                               port: int, 
                               api_key: str, 
                               path: str, 
                               query: dict,
                               index: str,
                               size: int,
                               fields: List):

    es_path = host + ":" + str(port) + path +"?index="+index+"&"+str(size)+"&pretty"
    es_header = "Authorization: ApiKey" + " " + api_key
    es_dict = {"query": query, "fields": fields}
    es_json = json.dumps(es_dict)
    es_replaced = " "+"'"+ es_json.replace("'", "\"").strip()+"'"

    result = run(["curl", "-k", "-XGET", "-H", "Content-Type: application/json", "-H",
                  es_header,
                  es_path,
                  "-d",
                  es_replaced]
                 # ,stdout=PIPE, stderr=PIPE, universal_newlines=True, shell=False
                )
    print(es_path, es_header, es_replaced)
    print(result.stdout)


elasticsearch_search_query(host="https://internal-a02c1b85dade940ef871b0b6d1e65191-1270244841.us-west-2.elb.amazonaws.com:9200/_search", api_key="Q2txZ2FvSUJ6Nlcwa3pjbnh0NUM6enBidjdNLTdRQVNhc0tuTEk5UEZmZw==",
                           port=9200, size=5, query={"match": {"_id": "c8a537a9-7f98-489b-8f8b-de2450c67ce6"}}, fields=["_id", "_index"], path="/_search",index=".fleet-enrollment-api-keys-7")

报错信息 #

{
  "error" : {
    "root_cause" : [
      {
        "type" : "illegal_argument_exception",
        "reason" : "request [/_search:9200/_search] contains unrecognized parameter: [5]"
      }
    ],
    "type" : "illegal_argument_exception",
    "reason" : "request [/_search:9200/_search] contains unrecognized parameter: [5]"
  },
  "status" : 400
}

https://internal-a02c1b85dade940ef871b0b6d1e65191-1270244841.us-west-2.elb.amazonaws.com:9200/_search:9200/_search?index=.fleet-enrollment-api-keys-7&5&pretty Authorization: ApiKey Q2txZ2FvSUJ6Nlcwa3pjbnh0NUM6enBidjdNLTdRQVNhc0tuTEk5UEZmZw== '{"query": {"match": {"_id": "c8a537a9-7f98-489b-8f8b-de2450c67ce6"}}, "fields": ["_id", "_index"]}'
None

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   350  100   350    0     0  10294      0 --:--:-- --:--:-- --:--:-- 10294
curl: (3) nested brace in URL position 12:
'{"query": {"match": {"_id": "c8a537a9-7f98-489b-8f8b-de2450c67ce6"}}, "fields": ["_id", "_index"]}'

更新：进程参数详情 #

CompletedProcess(args=['curl', '-k', '-XGET', '-H', 'Content-Type: application/json', '-H', 'Authorization: ApiKey Q2txZ2FvSUJ6Nlcwa3pjbnh0NUM6enBidjdNLTdRQVNhc0tuTEk5UEZmZw==', 'https://internal-a02c1b85dade940ef871b0b6d1e65191-1270244841.us-west-2.elb.amazonaws.com:9200:9200/_search?index=.fleet-enrollment-api-keys-7&size=5&pretty', '-d', '{"query": {"match": {"_id": "c8a537a9-7f98-489b-8f8b-de2450c67ce6"}}, "fields": ["_id", "_index"]}'], returncode=3)

问题分析与修复方案 #

问题1：URL拼接重复且参数缺失键名 #

从进程参数可见，URL出现9200:9200重复，因为传入的host已包含https://xxx:9200/_search，但代码额外拼接了:9200/_search，导致路径重复。同时size参数仅拼接了&5，缺少size=前缀，Elasticsearch无法识别该参数。

问题2：-d参数手动添加多余单引号 #

subprocess.run会自动处理参数边界，无需手动给JSON数据加单引号。额外添加的单引号会被curl当作请求数据的一部分，引发解析错误。

修复后的代码 #

import json
from builtins import int
from pydantic import BaseModel, Field
from typing import List, Dict
from subprocess import PIPE, run

def elasticsearch_search_query(host: str, 
                               port: int, 
                               api_key: str, 
                               path: str, 
                               query: dict,
                               index: str,
                               size: int,
                               fields: List):

    # 修复URL：利用f-string拼接，避免重复路径和参数缺失
    es_path = f"{host}?index={index}&size={size}&pretty"
    es_header = f"Authorization: ApiKey {api_key}"
    es_dict = {"query": query, "fields": fields}
    es_json = json.dumps(es_dict)
    # 直接传递JSON字符串，去掉多余单引号
    es_data = es_json.strip()

    # 开启stdout/stderr捕获，方便排查问题
    result = run(["curl", "-k", "-XGET", "-H", "Content-Type: application/json", "-H",
                  es_header,
                  es_path,
                  "-d",
                  es_data],
                 stdout=PIPE, stderr=PIPE, universal_newlines=True)
    print(es_path, es_header, es_data)
    print("STDOUT:", result.stdout)
    print("STDERR:", result.stderr)


elasticsearch_search_query(host="https://internal-a02c1b85dade940ef871b0b6d1e65191-1270244841.us-west-2.elb.amazonaws.com:9200/_search", api_key="Q2txZ2FvSUJ6Nlcwa3pjbnh0NUM6enBidjdNLTdRQVNhc0tuTEk5UEZmZw==",
                           port=9200, size=5, query={"match": {"_id": "c8a537a9-7f98-489b-8f8b-de2450c67ce6"}}, fields=["_id", "_index"], path="/_search",index=".fleet-enrollment-api-keys-7")

额外建议 #

优先使用Python官方Elasticsearch客户端（如elasticsearch-py）替代subprocess调用curl，代码更简洁易维护，还能避免命令行参数拼接的各类问题。

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