fixed json handling issue

.github/workflows/test.yaml

@@ -22,6 +22,8 @@ jobs:
         exclude:
           - os: macos-latest
             python-version: "3.8"
+          - os: macos-latest
+            python-version: "3.13"            
     runs-on: ${{matrix.os}}
     name: 'Run Tests on ${{matrix.os}} with Python ${{matrix.python-version}}'
 

CHANGELOG.md

@@ -1,11 +1,15 @@
 # Changelog
 
-## v3.7.1 (2024-11-19)
+## v3.7.2 (2024-11-19)
 
 ### Updates
 
 - Added `http_debug` constructor argument to return HTTP log information
 
+### Bug Fixes
+
+- Fixed issue passing JSON strings to queries, added test
+
 ## v3.7.0 (2024-11-08)
 
 ### Updates

pystackql/stackql.py

@@ -205,8 +205,20 @@ def _run_query(self, query, custom_auth=None, env_vars=None):
 		:raises FileNotFoundError: If the StackQL binary isn't found.
 		:raises Exception: For any other exceptions during the execution, providing a generic error message.
 		"""
+
 		local_params = self.params.copy()
-		local_params.insert(1, f'"{query}"')
+		script_path = None
+
+		if self.platform.startswith("Windows"):
+			# Escape double quotes and wrap in double quotes for Windows
+			escaped_query = query.replace('"', '\\"')  # Escape double quotes properly
+			safe_query = f'"{escaped_query}"'
+		else:
+			# Use shlex.quote for Unix-like systems
+			import shlex
+			safe_query = shlex.quote(query)
+
+		local_params.insert(1, safe_query)		
 		script_path = None
 
 		# Handle custom authentication if provided
@@ -240,19 +252,32 @@ def _run_query(self, query, custom_auth=None, env_vars=None):
 			full_command = " ".join([self.bin_path] + local_params)
 
 		try:
-			with subprocess.Popen(full_command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as iqlPopen:
-				stdout, stderr = iqlPopen.communicate()
-
-				if self.debug:
-					self._debug_log(f"query: {query}")
-					self._debug_log(f"stdout: {stdout}")
-					self._debug_log(f"stderr: {stderr}")
-
-				# Process stdout and stderr
-				if stderr:
-					output["error"] = stderr.decode('utf-8') if isinstance(stderr, bytes) else str(stderr)
-				if stdout:
-					output["data"] = stdout.decode('utf-8') if isinstance(stdout, bytes) else str(stdout)
+
+			full_command = full_command.replace("\n", " ")
+
+			result = subprocess.run(
+				full_command,
+				shell=True,
+				text=True,
+				capture_output=True
+			)
+
+			stdout = result.stdout
+			stderr = result.stderr
+			returncode = result.returncode
+
+			if self.debug:
+				self._debug_log(f"fullcommand: {full_command}")
+				self._debug_log(f"returncode: {returncode}")
+				self._debug_log(f"stdout: {stdout}")
+				self._debug_log(f"stderr: {stderr}")
+
+			# Process stdout and stderr
+			if stderr:
+				output["error"] = stderr.decode('utf-8') if isinstance(stderr, bytes) else str(stderr)
+			if stdout:
+				output["data"] = stdout.decode('utf-8') if isinstance(stdout, bytes) else str(stdout)
+
 
 		except FileNotFoundError:
 			output["exception"] = f"ERROR: {self.bin_path} not found"

setup.py

@@ -10,7 +10,7 @@
 
 setup(
     name='pystackql',
-    version='v3.7.1',
+    version='v3.7.2',
     description='A Python interface for StackQL',
     long_description=readme,
     author='Jeffrey Aven',

tests/pystackql_tests.py

@@ -263,6 +263,20 @@ def test_18_execute_custom_auth_env_vars(self):
         self.assertEqual(result, expected_result, f"Expected result: {expected_result}, got: {result}")
         print_test_result(f"Test 18 execute with custom auth and command-specific environment variables\nRESULT: {result}", result == expected_result)
 
+    @pystackql_test_setup()
+    def test_19_json_extract_function(self):
+        query = """
+        SELECT
+        json_extract('{"Key":"StackName","Value":"aws-stack"}', '$.Key') as key,
+        json_extract('{"Key":"StackName","Value":"aws-stack"}', '$.Value') as value
+        """
+        expected_result = [
+            {'key': {'String': 'StackName', 'Valid': True}, 'value': {'String': 'aws-stack', 'Valid': True}}
+        ]
+        result = self.stackql.execute(query)
+        self.assertEqual(result, expected_result, f"Expected result: {expected_result}, got: {result}")
+        print_test_result(f"Test 19 complex object handling\nRESULT: {result}", result == expected_result)
+
 
 @unittest.skipIf(platform.system() == "Windows", "Skipping async tests on Windows")
 class PyStackQLAsyncTests(PyStackQLTestsBase):