Splunk Addon Python模块TypeError问题排查及输出实现

算力阿峰

2026-6-25

解决Splunk模块化输入调用Sinch API的TypeError问题

问题背景

编写了用于Splunk模块化输入的Python代码，目标是调用Sinch API获取群组数据，但运行时抛出TypeError。

原代码

# encoding = utf-8

import os
import sys
import time
import datetime
import json

'''
    IMPORTANT
    Edit only the validate_input and collect_events functions.
    Do not edit any other part in this file.
    This file is generated only once when creating the modular input.
'''
'''
# For advanced users, if you want to create single instance mod input, uncomment this method.
def use_single_instance_mode():
    return True
'''

def validate_input(helper, definition):
    """Implement your own validation logic to validate the input stanza configurations"""
    # This example accesses the modular input variable
    # service_plan_id = definition.parameters.get('service_plan_id', None)
    # api_key = definition.parameters.get('api_key', None)
    pass

def collect_events(helper, ew):
    

    # The following examples get the arguments of this input.
    # Note, for single instance mod input, args will be returned as a dict.
    # For multi instance mod input, args will be returned as a single value.
    global opt_service_plan_id
    global opt_api_key
    opt_service_plan_id = helper.get_arg('service_plan_id')
    opt_api_key = helper.get_arg('api_key')
    
    service_plan_id = opt_service_plan_id
    region = "us"
    api_key = opt_api_key
    url = "https://" + region + ".sms.api.sinch.com/xms/v1/" + service_plan_id + "/groups"
    headers = {
        "Authorization": "Bearer {}".format(api_key)}
    response = helper.send_http_request(url, 'GET', parameters=None, payload=None,
                                        headers=headers, cookies=None, verify=True, cert=None,
                                        timeout=None, use_proxy=True)
    # get the response headers
    # r_headers = response.headers
    # get the response body as text
    # r_text = response.text
    # get response body as json. If the body text is not a json string, raise a ValueError
    r_json = response.json()
    # get response cookies
    # r_cookies = response.cookies
    # get redirect history
    # historical_responses = response.history
    # get response status code
    r_status = response.status_code
    if r_status !=200:
       # check the response status, if the status is not sucessful, raise requests.HTTPError
       response.raise_for_status()

    # The following examples show usage of check pointing related helper functions.
    # save checkpoint
    #helper.save_check_point(key, state)
    # delete checkpoint
    #helper.delete_check_point(key)
    # get checkpoint
    #state = helper.get_check_point(key)

    # To create a splunk event
    event = helper.new_event(json.dumps(r_json['groups']), time=None, host=None, index=None, source=None, sourcetype=None, done=True, unbroken=True)
    ew.write_event(event)

错误信息

Traceback (most recent call last):
  File "/opt/splunk/etc/apps/TA-sinch-addon-for-splunk/bin/ta_sinch_addon_for_splunk/aob_py3/modinput_wrapper/base_modinput.py", line 128, in stream_events
    self.collect_events(ew)
  File "/opt/splunk/etc/apps/TA-sinch-addon-for-splunk/bin/sinchpythonaddontest_1713375439_287.py", line 64, in collect_events
    input_module.collect_events(self, ew)
  File "/opt/splunk/etc/apps/TA-sinch-addon-for-splunk/bin/input_module_sinchpythonaddontest_1713375439_287.py", line 83, in collect_events
    url = "https://" + region + ".sms.api.sinch.com/xms/v1/" + service_plan_id + "/groups"
TypeError: can only concatenate str (not "NoneType") to str

错误原因

拼接URL时service_plan_id为None，无法与字符串进行拼接。同时原代码未对输入参数做校验，也未输出包含page_size的完整JSON结构。

预期输出

需要输出包含群组列表及page_size的JSON数据：

{
  "groups": [
    {
      "id": "01HTCSYHRFC7Z7PZ3KJMQC8Y56",
      "name": "Rohit9",
      "size": 2,
      "created_at": "2024-04-01T12:25:36.271Z",
      "modified_at": "2024-04-01T12:25:36.271Z",
      "child_groups": []
    },
    {
      "id": "01HT1TJ2EK6T4N0KC4PDJV4K8H",
      "name": "Rohit3",
      "size": 1,
      "created_at": "2024-03-28T06:04:37.203Z",
      "modified_at": "2024-03-28T06:04:37.203Z",
      "child_groups": []
    },
    {
      "id": "01HT1T90Q5WT5SNM4S253RYS00",
      "name": "Rohit3",
      "size": 0,
      "created_at": "2024-03-28T05:59:40.517Z",
      "modified_at": "2024-03-28T05:59:40.517Z",
      "child_groups": []
    },
    {
      "id": "01HT1T7RB9K15HXNCNFFJ4AA67",
      "name": "YOUR_group_name",
      "size": 0,
      "created_at": "2024-03-28T05:58:59.177Z",
      "modified_at": "2024-03-28T05:58:59.177Z",
      "child_groups": []
    },
    {
      "id": "01HT0YR44GGQTBR21J5KCR8854",
      "name": "Rohit4",
      "size": 1,
      "created_at": "2024-03-27T21:58:35.408Z",
      "modified_at": "2024-03-27T21:58:35.408Z",
      "child_groups": []
    },
    {
      "id": "01HT0YPPNKR567KBDB0YS8CG0B",
      "name": "Rohit4",
      "size": 1,
      "created_at": "2024-03-27T21:57:48.851Z",
      "modified_at": "2024-03-27T21:57:48.851Z",
      "child_groups": []
    },
    {
      "id": "01HT0YNF0MTYBWMY9B93ECYSHC",
      "name": "Rohit4",
      "size": 1,
      "created_at": "2024-03-27T21:57:08.244Z",
      "modified_at": "2024-03-27T21:57:08.244Z",
      "child_groups": []
    },
    {
      "id": "01HT0W025PM8G120QA1Q73SKP0",
      "name": "Rohit4",
      "size": 1,
      "created_at": "2024-03-27T21:10:29.814Z",
      "modified_at": "2024-03-27T21:10:29.814Z",
      "child_groups": []
    },
    {
      "id": "01HT0VFBX0Y9BPHYNBRD6SHQ39",
      "name": "Rohit3",
      "size": 2,
      "created_at": "2024-03-27T21:01:22.720Z",
      "modified_at": "2024-03-27T21:01:22.720Z",
      "child_groups": []
    },
    {
      "id": "01HT0V0C5VB21YEYK6JPN69YQK",
      "name": "Rohit2",
      "size": 2,
      "created_at": "2024-03-27T20:53:11.483Z",
      "modified_at": "2024-03-27T20:53:11.483Z",
      "child_groups": []
    },
    {
      "id": "01HSJS9TS1W8RJ951H0AVKH2QC",
      "name": "RohitpandeyGroup",
      "size": 2,
      "created_at": "2024-03-22T09:54:02.145Z",
      "modified_at": "2024-03-22T09:54:02.145Z",
      "child_groups": []
    }
  ],
  "page_size": 11
}

修复方案

1. 完善参数校验逻辑

在validate_input函数中添加参数校验，确保service_plan_id和api_key不为空且是有效字符串，提前拦截无效配置。

2. 优化URL拼接方式

使用f-string格式化URL，更简洁且避免类型拼接错误，同时去掉不必要的全局变量。

3. 输出完整JSON结构

修改事件生成逻辑，将包含groups和page_size的完整响应JSON输出，而不是仅输出groups列表。

修复后的代码

# encoding = utf-8

import os
import sys
import time
import datetime
import json

'''
    IMPORTANT
    Edit only the validate_input and collect_events functions.
    Do not edit any other part in this file.
    This file is generated only once when creating the modular input.
'''
'''
# For advanced users, if you want to create single instance mod input, uncomment this method.
def use_single_instance_mode():
    return True
'''

def validate_input(helper, definition):
    """Implement your own validation logic to validate the input stanza configurations"""
    service_plan_id = definition.parameters.get('service_plan_id')
    api_key = definition.parameters.get('api_key')
    
    # 校验service_plan_id非空且为有效字符串
    if not service_plan_id or not isinstance(service_plan_id, str) or service_plan_id.strip() == "":
        raise ValueError("service_plan_id must be a non-empty string")
    
    # 校验api_key非空且为有效字符串
    if not api_key or not isinstance(api_key, str) or api_key.strip() == "":
        raise ValueError("api_key must be a non-empty string")

def collect_events(helper, ew):
    # 获取输入参数
    service_plan_id = helper.get_arg('service_plan_id').strip()
    api_key = helper.get_arg('api_key').strip()
    region = "us"
    
    # 使用f-string安全拼接URL
    url = f"https://{region}.sms.api.sinch.com/xms/v1/{service_plan_id}/groups"
    
    headers = {
        "Authorization": f"Bearer {api_key}"
    }
    
    response = helper.send_http_request(url, 'GET', parameters=None, payload=None,
                                        headers=headers, cookies=None, verify=True, cert=None,
                                        timeout=None, use_proxy=True)
    
    # 校验响应状态码
    if response.status_code != 200:
        response.raise_for_status()
    
    r_json = response.json()
    
    # 生成包含groups和page_size的完整事件数据
    event_data = json.dumps({
        "groups": r_json.get('groups', []),
        "page_size": r_json.get('page_size', 0)
    })
    
    # 创建并写入Splunk事件
    event = helper.new_event(event_data, time=None, host=None, index=None, source=None, sourcetype=None, done=True, unbroken=True)
    ew.write_event(event)

验证说明

参数校验：当service_plan_id或api_key为空时，在输入配置阶段就会抛出错误，避免进入执行逻辑后出现TypeError。
URL拼接：使用f-string确保所有变量都是字符串类型（已通过校验），不会出现None拼接问题。
输出结构：事件数据包含groups列表和page_size字段，完全符合预期输出格式。

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

Splunk Addon Python模块TypeError问题排查及输出实现

问题背景 #

原代码 #

错误信息 #

错误原因 #

预期输出 #

修复方案 #

1. 完善参数校验逻辑 #

2. 优化URL拼接方式 #

3. 输出完整JSON结构 #

修复后的代码 #

验证说明 #