Skip to content

Instantly share code, notes, and snippets.

@marcusschiesser
Created March 16, 2023 09:15
Show Gist options
  • Save marcusschiesser/ee2c1737b31d0d1f927fa7924d6cda33 to your computer and use it in GitHub Desktop.
Save marcusschiesser/ee2c1737b31d0d1f927fa7924d6cda33 to your computer and use it in GitHub Desktop.
JSON handlers for Splunk Custom REST endpoints
import logging
import traceback
import json
from typing import Callable
def json_handler(func: Callable):
def wrapper(*args, **kwargs):
decorated = json_exception_handler(json_payload_extractor(func))
return decorated(*args, **kwargs)
return wrapper
def convert_tuples_to_dict(tuples):
return {t[0]: t[1] for t in tuples}
def json_payload_extractor(func: Callable):
def wrapper(self, in_string):
try:
request = json.loads(in_string)
kwargs = {"request": request, "in_string": in_string}
if "payload" in request:
# if request contains payload, parse it and add it as payload parameter
kwargs["payload"] = json.loads(request["payload"])
if "query" in request:
# if request contains query, parse it and add it as query parameter
kwargs["query"] = convert_tuples_to_dict(request["query"])
return func(self, **kwargs)
except ValueError as e:
return {
"payload": {
"success": "false",
"result": f"Error parsing JSON: {e}",
},
"status": 400,
}
return wrapper
def json_exception_handler(func):
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
logging.error(f"error={repr(e)} traceback={traceback.format_exc()}")
return {
"payload": {"success": "false", "message": f"Error: {repr(e)}"},
"status": 500,
}
return wrapper
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment