如何用Python执行指定curl命令，将SQL结果转JSON推送到New Relic

Got it, let's walk through how to translate that curl command into Python, plus integrate SQL query results into the mix. Here's a step-by-step solution:

Prerequisites #

First, install the required packages. You'll need requests for making HTTP calls, plus a database connector for your SQL database (e.g., psycopg2 for PostgreSQL, mysql-connector-python for MySQL, sqlite3 is built-in for SQLite).

pip install requests psycopg2-binary  # Replace psycopg2-binary with your DB driver if needed

Step 1: Fetch SQL Query Results #

Let's write code to connect to your database, run a query, and convert the results into a list of dictionaries (easy to turn into JSON later). I'll use PostgreSQL as an example, but the pattern works for other databases with minor tweaks.

import psycopg2  # Replace with your DB connector (e.g., mysql.connector, sqlite3)

# Database connection details - update these with your own
DB_HOST = "your_db_host"
DB_NAME = "your_db_name"
DB_USER = "your_db_user"
DB_PASSWORD = "your_db_password"

def fetch_sql_results():
    conn = None
    try:
        # Connect to the database
        conn = psycopg2.connect(
            host=DB_HOST,
            database=DB_NAME,
            user=DB_USER,
            password=DB_PASSWORD
        )
        
        # Run your SQL query
        query = "SELECT attribute1, attribute2, attribute3 FROM your_table WHERE some_condition;"
        cursor = conn.cursor()
        cursor.execute(query)
        
        # Convert rows to dictionaries (maps column names to values)
        columns = [desc[0] for desc in cursor.description]
        results = [dict(zip(columns, row)) for row in cursor.fetchall()]
        
        return results
    except Exception as e:
        print(f"Error fetching SQL results: {e}")
        return []
    finally:
        if conn:
            conn.close()

# Get your query results
sql_results = fetch_sql_results()

Step 2: Format Data for New Relic #

New Relic expects events to have an eventType field, plus any custom attributes. We'll wrap each SQL result row into an event object:

# Define your custom event type (use underscores instead of spaces for best practices)
EVENT_TYPE = "Custom_Event_Name"

# Create a list of New Relic events
new_relic_events = []
for row in sql_results:
    event = {
        "eventType": EVENT_TYPE,
        **row  # Unpack all attributes from the SQL row into the event
    }
    new_relic_events.append(event)

Step 3: Send Data to New Relic via POST #

Now we'll replicate the curl command using Python's requests library. This is where we send the formatted events to New Relic's Insights API:

import requests

# New Relic configuration - update these with your actual values
NEW_RELIC_ACCOUNT_ID = "YOUR_ACCOUNT_ID"
NEW_RELIC_INSERT_KEY = "YOUR_KEY_HERE"
NEW_RELIC_API_URL = f"https://insights-collector.newrelic.com/v1/accounts/{NEW_RELIC_ACCOUNT_ID}/events"

def send_to_new_relic(events):
    if not events:
        print("No events to send.")
        return
    
    headers = {
        "Content-Type": "application/json",
        "X-Insert-Key": NEW_RELIC_INSERT_KEY
    }
    
    try:
        response = requests.post(NEW_RELIC_API_URL, headers=headers, json=events)
        response.raise_for_status()  # Raise an error for HTTP status codes >=400
        print(f"Successfully sent {len(events)} events to New Relic!")
    except requests.exceptions.RequestException as e:
        print(f"Failed to send events: {e}")
        if hasattr(e, 'response') and e.response:
            print(f"Response content: {e.response.text}")

# Send the events
send_to_new_relic(new_relic_events)

Key Notes to Keep in Mind #

• Batch Size: New Relic allows up to 1000 events per request. If your SQL query returns more than that, split the new_relic_events list into chunks of 1000 before sending.
• Event Naming: Avoid spaces in eventType—use underscores instead (e.g., User_Login_Event instead of User Login Event) to avoid issues in New Relic's UI.
• Error Handling: The example includes basic error handling, but you might want to add retries for transient errors (using something like tenacity or custom retry logic).
• Database Drivers: If you're using MySQL, replace psycopg2 with mysql-connector-python and adjust the connection code accordingly. For SQLite, use the built-in sqlite3 module.

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