Added parameters support

Author: Peng Ren

README.md

@@ -147,6 +147,40 @@ row = cursor.fetchone()
 print(row['name'])  # Access by column name
 ```
 
+### Query with Parameters
+
+PyMongoSQL supports two styles of parameterized queries for safe value substitution:
+
+**Positional Parameters with ?**
+
+```python
+from pymongosql import connect
+
+connection = connect(host="mongodb://localhost:27017/database")
+cursor = connection.cursor()
+
+cursor.execute(
+    'SELECT name, email FROM users WHERE age > ? AND status = ?',
+    [25, 'active']
+)
+```
+
+**Named Parameters with :name**
+
+```python
+from pymongosql import connect
+
+connection = connect(host="mongodb://localhost:27017/database")
+cursor = connection.cursor()
+
+cursor.execute(
+    'SELECT name, email FROM users WHERE age > :age AND status = :status',
+    {'age': 25, 'status': 'active'}
+)
+```
+
+Parameters are substituted into the MongoDB filter during execution, providing protection against injection attacks.
+
 ## Supported SQL Features
 
 ### SELECT Statements
@@ -201,8 +235,6 @@ This allows seamless integration between MongoDB data and Superset's BI capabili
 
 **Note**: Currently PyMongoSQL focuses on Data Query Language (DQL) operations. The following SQL features are **not yet supported** but are planned for future releases:
 
-- **Parameterized Queries**
-  - Parameter substitution support (?, :pram, etc.)
 - **DML Operations** (Data Manipulation Language)
   - `INSERT`, `UPDATE`, `DELETE`
 - **DDL Operations** (Data Definition Language)  

pymongosql/common.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 import logging
 from abc import ABCMeta, abstractmethod
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Tuple, Union
 
 from .error import ProgrammingError
 
@@ -38,12 +38,12 @@ def description(
     def execute(
         self,
         operation: str,
-        parameters: Optional[Dict[str, Any]] = None,
+        parameters: Optional[Union[Sequence[Any], Dict[str, Any]]] = None,
     ):
         raise NotImplementedError  # pragma: no cover
 
     @abstractmethod
-    def executemany(self, operation: str, seq_of_parameters: List[Optional[Dict[str, Any]]]) -> None:
+    def executemany(self, operation: str, seq_of_parameters: List[Union[Sequence[Any], Dict[str, Any]]]) -> None:
         raise NotImplementedError  # pragma: no cover
 
     @abstractmethod

pymongosql/cursor.py

@@ -75,21 +75,20 @@ def _check_closed(self) -> None:
         if self._is_closed:
             raise ProgrammingError("Cursor is closed")
 
-    def execute(self: _T, operation: str, parameters: Optional[Dict[str, Any]] = None) -> _T:
+    def execute(self: _T, operation: str, parameters: Optional[Any] = None) -> _T:
         """Execute a SQL statement
 
         Args:
             operation: SQL statement to execute
-            parameters: Parameters for the SQL statement (not yet implemented)
+            parameters: Parameters to substitute placeholders in the SQL
+                - Sequence for positional parameters with ? placeholders
+                - Dict for named parameters with :name placeholders
 
         Returns:
             Self for method chaining
         """
         self._check_closed()
 
-        if parameters:
-            _logger.warning("Parameter substitution not yet implemented, ignoring parameters")
-
         try:
             # Create execution context
             context = ExecutionContext(operation, self.mode)
@@ -98,7 +97,7 @@ def execute(self: _T, operation: str, parameters: Optional[Dict[str, Any]] = Non
             strategy = ExecutionPlanFactory.get_strategy(context)
 
             # Execute using selected strategy (Standard or Subquery)
-            result = strategy.execute(context, self.connection)
+            result = strategy.execute(context, self.connection, parameters)
 
             # Store execution plan for reference
             self._current_execution_plan = strategy.execution_plan

pymongosql/executor.py

@@ -2,7 +2,7 @@
 import logging
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, Sequence, Union
 
 from pymongo.errors import PyMongoError
 
@@ -19,6 +19,7 @@ class ExecutionContext:
 
     query: str
     execution_mode: str = "standard"
+    parameters: Optional[Union[Sequence[Any], Dict[str, Any]]] = None
 
     def __repr__(self) -> str:
         return f"ExecutionContext(mode={self.execution_mode}, " f"query={self.query})"
@@ -38,13 +39,15 @@ def execute(
         self,
         context: ExecutionContext,
         connection: Any,
+        parameters: Optional[Union[Sequence[Any], Dict[str, Any]]] = None,
     ) -> Optional[Dict[str, Any]]:
         """
         Execute query and return result set.
 
         Args:
             context: ExecutionContext with query and subquery info
             connection: MongoDB connection
+            parameters: Sequence for positional (?) or Dict for named (:param) parameters
 
         Returns:
             command_result with query results
@@ -86,19 +89,59 @@ def _parse_sql(self, sql: str) -> ExecutionPlan:
             _logger.error(f"SQL parsing failed: {e}")
             raise SqlSyntaxError(f"Failed to parse SQL: {e}")
 
-    def _execute_execution_plan(self, execution_plan: ExecutionPlan, db: Any) -> Optional[Dict[str, Any]]:
+    def _replace_placeholders(self, obj: Any, parameters: Sequence[Any]) -> Any:
+        """Recursively replace ? placeholders with parameter values in filter/projection dicts"""
+        param_index = [0]  # Use list to allow modification in nested function
+
+        def replace_recursive(value: Any) -> Any:
+            if isinstance(value, str):
+                # Replace ? with the next parameter value
+                if value == "?":
+                    if param_index[0] < len(parameters):
+                        result = parameters[param_index[0]]
+                        param_index[0] += 1
+                        return result
+                    else:
+                        raise ProgrammingError(
+                            f"Not enough parameters provided: expected at least {param_index[0] + 1}"
+                        )
+                return value
+            elif isinstance(value, dict):
+                return {k: replace_recursive(v) for k, v in value.items()}
+            elif isinstance(value, list):
+                return [replace_recursive(item) for item in value]
+            else:
+                return value
+
+        return replace_recursive(obj)
+
+    def _execute_execution_plan(
+        self,
+        execution_plan: ExecutionPlan,
+        db: Any,
+        parameters: Optional[Sequence[Any]] = None,
+    ) -> Optional[Dict[str, Any]]:
         """Execute an ExecutionPlan against MongoDB using db.command"""
         try:
             # Get database
             if not execution_plan.collection:
                 raise ProgrammingError("No collection specified in query")
 
+            # Replace placeholders with parameters in filter_stage only (not in projection)
+            filter_stage = execution_plan.filter_stage or {}
+
+            if parameters:
+                # Positional parameters with ? (named parameters are converted to positional in execute())
+                filter_stage = self._replace_placeholders(filter_stage, parameters)
+
+            projection_stage = execution_plan.projection_stage or {}
+
             # Build MongoDB find command
-            find_command = {"find": execution_plan.collection, "filter": execution_plan.filter_stage or {}}
+            find_command = {"find": execution_plan.collection, "filter": filter_stage}
 
             # Apply projection if specified
-            if execution_plan.projection_stage:
-                find_command["projection"] = execution_plan.projection_stage
+            if projection_stage:
+                find_command["projection"] = projection_stage
 
             # Apply sort if specified
             if execution_plan.sort_stage:
@@ -135,14 +178,28 @@ def execute(
         self,
         context: ExecutionContext,
         connection: Any,
+        parameters: Optional[Union[Sequence[Any], Dict[str, Any]]] = None,
     ) -> Optional[Dict[str, Any]]:
         """Execute standard query directly against MongoDB"""
         _logger.debug(f"Using standard execution for query: {context.query[:100]}")
 
+        # Preprocess query to convert named parameters to positional
+        processed_query = context.query
+        processed_params = parameters
+        if isinstance(parameters, dict):
+            # Convert :param_name to ? for parsing
+            import re
+
+            param_names = re.findall(r":(\w+)", context.query)
+            # Convert dict parameters to list in order of appearance
+            processed_params = [parameters[name] for name in param_names]
+            # Replace :param_name with ?
+            processed_query = re.sub(r":(\w+)", "?", context.query)
+
         # Parse the query
-        self._execution_plan = self._parse_sql(context.query)
+        self._execution_plan = self._parse_sql(processed_query)
 
-        return self._execute_execution_plan(self._execution_plan, connection.database)
+        return self._execute_execution_plan(self._execution_plan, connection.database, processed_params)
 
 
 class ExecutionPlanFactory:

pymongosql/superset_mongodb/executor.py

@@ -44,6 +44,7 @@ def execute(
         self,
         context: ExecutionContext,
         connection: Any,
+        parameters: Optional[Any] = None,
     ) -> Optional[Dict[str, Any]]:
         """Execute query in two stages: MongoDB for subquery, intermediate DB for outer query"""
         _logger.debug(f"Using subquery execution for query: {context.query[:100]}")
@@ -54,7 +55,7 @@ def execute(
         # If no subquery detected, fall back to standard execution
         if not query_info.has_subquery:
             _logger.debug("No subquery detected, falling back to standard execution")
-            return super().execute(context, connection)
+            return super().execute(context, connection, parameters)
 
         # Stage 1: Execute MongoDB subquery
         mongo_query = query_info.subquery_text

tests/test_cursor.py

@@ -398,3 +398,29 @@ def test_execute_with_nested_field_alias(self, conn):
         rows = cursor.result_set.fetchall()
         assert len(rows) == 3
         assert len(rows[0]) == 2  # Should have 2 columns
+
+    def test_execute_with_positional_parameters(self, conn):
+        """Test executing SELECT with positional parameters (?)"""
+        sql = "SELECT name, email FROM users WHERE age > ? AND active = ?"
+        cursor = conn.cursor()
+        result = cursor.execute(sql, [25, True])
+
+        assert result == cursor  # execute returns self
+        assert isinstance(cursor.result_set, ResultSet)
+
+        rows = cursor.result_set.fetchall()
+        assert len(rows) > 0  # Should have results matching the filter
+        assert len(rows[0]) == 2  # Should have name and email columns
+
+    def test_execute_with_named_parameters(self, conn):
+        """Test executing SELECT with named parameters (:name)"""
+        sql = "SELECT name, email FROM users WHERE age > :min_age AND active = :is_active"
+        cursor = conn.cursor()
+        result = cursor.execute(sql, {"min_age": 25, "is_active": True})
+
+        assert result == cursor  # execute returns self
+        assert isinstance(cursor.result_set, ResultSet)
+
+        rows = cursor.result_set.fetchall()
+        assert len(rows) > 0  # Should have results matching the filter
+        assert len(rows[0]) == 2  # Should have name and email columns