Method execute_query

camel/toolkits/sql_toolkit.py:257–385

r"""Execute a SQL query and return results. This method executes a SQL query against the configured database and returns the results. For SELECT queries, returns a list of dictionaries where each dictionary represents a row. For write operations (INSERT, UPDATE, DELE

(
        self,
        query: str,
        params: Optional[
            Union[
                List[Union[str, int, float, bool, None]],
                Dict[str, Union[str, int, float, bool, None]],
            ]
        ] = None,
    )

        return f'"{escaped}"'

    def execute_query(
        self,
        query: str,
        params: Optional[
            Union[
                List[Union[str, int, float, bool, None]],
                Dict[str, Union[str, int, float, bool, None]],
            ]
        ] = None,
    ) -> Union[List[Dict[str, Any]], Dict[str, Any], str]:
        r"""Execute a SQL query and return results.

        This method executes a SQL query against the configured database and
        returns the results. For SELECT queries, returns a list of dictionaries
        where each dictionary represents a row. For write operations (INSERT,
        UPDATE, DELETE, etc.), returns a status dictionary with execution info.

        Args:
            query (str): The SQL query to execute.
            params (Optional[Union[List[Union[str, int, float, bool, None]],
                Dict[str, Union[str, int, float, bool, None]]]], optional):
                Parameters for parameterized queries. Can be a list for
                positional parameters (with ? placeholders) or a dict for
                named parameters. Values can be strings, numbers, booleans,
                or None. Note: tuples are also accepted at runtime but should
                be passed as lists for type compatibility.
                (default: :obj:`None`)

        Returns:
            Union[List[Dict[str, Any]], Dict[str, Any], str]:
                - For SELECT queries: List of dictionaries with column names as
                  keys and row values as values.
                - For write operations (INSERT, UPDATE, DELETE, CREATE, etc.):
                  A dictionary with 'status', 'message', and optionally
                  'rows_affected' keys.
                - For errors: An error message string starting with "Error:".
        """
        if not query or not query.strip():
            return "Error: Query cannot be empty"

        # Validate query mode (check for write operations in read-only mode)
        if self.read_only and self._is_write_query(query):
            return (
                "Error: Write operations are not allowed in read-only mode. "
                "The query contains write operations (INSERT, UPDATE, DELETE, "
                "DROP, CREATE, ALTER, TRUNCATE, etc.). Only SELECT queries "
                "are permitted."
            )

        try:
            logger.debug(f"Executing query: {query}...")

            cursor = self._connection.cursor()

            # Execute query with or without parameters
            if params is not None:
                cursor.execute(query, params)
            else: