Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def run(self, params=None):
if params is None:
params = {}
json_in = params.get(Input.JSON_IN)
flags = params.get(Input.FLAGS)
filter_ = params.get(Input.FILTER)
timeout = params.get(Input.TIMEOUT)
jq_cmd_array = ["jq"]
if len(flags) > 0:
string_flags = ' '.join(flags)
jq_cmd_array.append(string_flags)
jq_cmd_array.append(filter_)
self.logger.info("Command to Run: {}".format(jq_cmd_array))
process = subprocess.Popen(jq_cmd_array, stdout=PIPE, stderr=PIPE, stdin=PIPE)
std_out, std_err = process.communicate(input=json.dumps(json_in).encode(), timeout=timeout)
return_code = process.returncode
self.logger.info("Return Code: {}".format(return_code))
if return_code > 0: