def _invoke()

in tools/hologres_excute_sql.py [0:0]

• 46 lines of code
• 15 McCabe index (conditional complexity)

    def _invoke(self, tool_parameters: dict[str, Any]) -> Generator[ToolInvokeMessage]:
        # Get the SQL statement passed in
        sql = tool_parameters.get("sql")
        if not sql:
            raise ValueError("SQL statement cannot be empty")
        
        # Improved risk detection
        if self._contains_risk_commands(sql):
            raise ValueError("SQL statement contains risks")

        # Get database connection parameters
        db_type = tool_parameters.get("db_type")
        host = tool_parameters.get("host")
        port = tool_parameters.get("port")
        database = tool_parameters.get("db_name")
        username = tool_parameters.get("username")
        password = tool_parameters.get("password")

        if not all([db_type, host, port, database, username, password]):
            raise ValueError("Database connection parameters cannot be empty")
        
        try:
            # Execute SQL statement (query or non-query)
            result = execute_sql(
                db_type, host, int(port), database, 
                username, password, sql, ""
            )
            # Handle empty results
            if isinstance(result, list) and not result:  # Empty query result
                yield self.create_text_message("No data found")
            elif isinstance(result, dict) and "rowcount" in result and result["rowcount"] == 0:  # No affected rows
                yield self.create_text_message("No data affected")

            result_format = tool_parameters.get("result_format", "json")
            
            if result_format == 'json':
                yield self.create_json_message({
                        "status": "success",
                        "result": result
                    }  
                )
            elif result_format == 'csv':
                yield from self._handle_csv(result)
            elif result_format == 'html':
                yield from self._handle_html(result)
            else:
                if result is not None:
                    message_text = json.dumps(
                        result, 
                        ensure_ascii=False, 
                        default=self._custom_serializer  # Key modification point
                    )
                else:
                    message_text = "No data found"
                yield self.create_text_message(message_text)
        except Exception as e:
            raise ValueError(f"Database operation failed: {str(e)}")