如何在Python版Google Cloud Function中使用Cloud Scheduler的Body参数

Let's fix this step by step. The main issues here are that your current code doesn't extract parameters from the POST request body, and we should also address a critical security risk with how you're building your query.

Step 1: Parse the POST Request Body #

Cloud Scheduler sends your parameters as a JSON payload, so we'll start by extracting and validating that data from the request object.

Step 2: Use Parameterized Queries (Avoid SQL Injection!) #

Never use string formatting to plug external values into SQL queries—this leaves you open to SQL injection attacks. Instead, use BigQuery's built-in parameterized query support for safe, reliable dynamic queries.

Modified Full Code #

Here's your updated function with all necessary changes:

import flask
from google.cloud import bigquery
from google.cloud.bigquery import ScalarQueryParameter

app = flask.Flask(__name__)

def main(request):
    with app.app_context():
        # Parse the JSON request body (silent=True returns None for invalid JSON)
        request_json = request.get_json(silent=True)
        
        # Validate required parameters exist to avoid unexpected errors
        if not request_json or 'unit' not in request_json or 'interval' not in request_json:
            return flask.jsonify({"error": "Missing required parameters: 'unit' and 'interval' are mandatory"}), 400
        
        # Extract the parameters from the parsed JSON
        unit = request_json['unit']
        interval = request_json['interval']
        
        # Use parameterized placeholders (@unit, @interval) instead of string formatting
        query = """
            SELECT unitId 
            FROM `myproject.mydataset.mytable` 
            WHERE unit = @unit 
              AND interval = @interval
        """
        
        client = bigquery.Client()
        job_config = bigquery.QueryJobConfig()
        
        # Define the query parameters with their data types (adjust type if interval is numeric)
        job_config.query_parameters = [
            ScalarQueryParameter("unit", "STRING", unit),
            ScalarQueryParameter("interval", "STRING", interval)
        ]
        
        # Configure destination table settings
        dest_dataset = client.dataset('mydataset', 'myproject')
        dest_table = dest_dataset.table('mytable')
        job_config.destination = dest_table
        job_config.create_disposition = 'CREATE_IF_NEEDED'
        job_config.write_disposition = 'WRITE_APPEND'
        
        # Execute the query and wait for completion
        job = client.query(query, job_config=job_config)
        job.result()
        
        return flask.jsonify({"status": "Success", "message": "Query completed and results written to table"}), 200

Important Setup Notes for Cloud Scheduler: #

• Ensure your job uses the POST method
• The request body must use valid JSON (only double quotes, like {"unit": "myunitname", "interval": "1"})
• Add a Content-Type header with value application/json to tell Cloud Function how to parse the payload

内容的提问来源于stack exchange，提问作者Kaustubh Ghole