如何用Python高效将HTTP响应存入SQLite3并处理时区转换？

Hey there! Since you're new to Python and tackling this project of building an SQL database for historical energy prices from a Web API, let's break this down into simple, actionable steps. I'll cover everything from fetching data to converting time zones and inserting into SQL.

First, grab the tools you'll need. Open your terminal and run these commands:

pip install requests pytz  # For API calls and time zone handling
pip install beautifulsoup4  # Only if you need to parse XML data
# Pick your SQL database package (example options below)
pip install sqlite3  # Built-in for lightweight testing
pip install psycopg2-binary  # For PostgreSQL
pip install pyodbc  # For SQL Server

Let's start with pulling data from the API. Most APIs return JSON, so we'll use that as the primary example—we'll cover XML too.

Example for JSON API: #

import requests

# Replace with your actual API details
api_url = "https://your-api-endpoint.com/energy-prices"
headers = {"Authorization": "Bearer YOUR_API_KEY"}  # Add if your API requires auth

try:
    response = requests.get(api_url, headers=headers)
    response.raise_for_status()  # Throw an error if the request fails
    raw_data = response.json()  # Convert JSON response to a Python list/dict
except requests.exceptions.RequestException as e:
    print(f"Oops, failed to fetch data: {e}")

Example for XML API: #

If the API returns XML, use BeautifulSoup to parse it into a usable format:

from bs4 import BeautifulSoup

# After fetching the response
xml_content = response.text
soup = BeautifulSoup(xml_content, "xml")

# Extract data (adjust based on your XML structure)
nodes = soup.find_all("node")
raw_data = []
for node in nodes:
    node_data = {
        "node_id": node.find("id").text,
        "price": float(node.find("price").text),
        "edt_timestamp": node.find("timestamp").text
    }
    raw_data.append(node_data)

Don't waste time manually checking daylight saving rules—let Python handle it automatically. Use zoneinfo (Python 3.9+) or pytz for older versions to work with the America/New_York timezone, which switches between EDT and EST on its own.

Using zoneinfo (Python 3.9+): #

from zoneinfo import ZoneInfo
from datetime import datetime

def convert_edt_to_est(edt_timestamp_str):
    # Parse the EDT string into a datetime object (adjust the format to match your data!)
    edt_dt = datetime.strptime(edt_timestamp_str, "%Y-%m-%d %H:%M:%S")
    # Mark the datetime as being in EDT (America/New_York timezone)
    edt_localized = edt_dt.replace(tzinfo=ZoneInfo("America/New_York"))
    # Convert to EST (the timezone will automatically adjust for DST)
    # Return a timezone-aware datetime for SQL (most databases support this)
    return edt_localized.astimezone(ZoneInfo("America/New_York"))

Using pytz (for older Python versions): #

import pytz
from datetime import datetime

def convert_edt_to_est(edt_timestamp_str):
    edt_dt = datetime.strptime(edt_timestamp_str, "%Y-%m-%d %H:%M:%S")
    eastern_tz = pytz.timezone("America/New_York")
    # Mark the datetime as EDT (daylight saving active)
    edt_localized = eastern_tz.localize(edt_dt, is_dst=True)
    # Convert to standard EST time
    return edt_localized.astimezone(eastern_tz)

Pro Tip: Always store timezone-aware datetimes in your SQL database if possible—it avoids confusion later when querying historical data.

Now process each entry to convert the timestamp and structure it for your database:

processed_data = []
for entry in raw_data:
    try:
        est_timestamp = convert_edt_to_est(entry["edt_timestamp"])
        processed_entry = (
            entry["node_id"],
            entry["price"],
            est_timestamp
        )
        processed_data.append(processed_entry)
    except KeyError as e:
        print(f"Missing data in entry: {e}—skipping this one")
        continue

Let's use SQLite as an example (it's built-in and great for testing). First create your table, then insert the processed data.

Example with SQLite: #

import sqlite3

# Connect to your database (creates it if it doesn't exist)
conn = sqlite3.connect("energy_prices.db")
cursor = conn.cursor()

# Create table (run this once)
cursor.execute("""
CREATE TABLE IF NOT EXISTS node_prices (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    node_id TEXT NOT NULL,
    price REAL NOT NULL,
    est_timestamp DATETIME NOT NULL,
    UNIQUE(node_id, est_timestamp)  # Prevent duplicate entries for the same node/time
)
""")

# Batch insert data (much faster than inserting one by one!)
try:
    cursor.executemany("""
    INSERT OR IGNORE INTO node_prices (node_id, price, est_timestamp)
    VALUES (?, ?, ?)
    """, processed_data)
    conn.commit()
    print(f"Successfully added {cursor.rowcount} entries to the database!")
except sqlite3.Error as e:
    print(f"Database error: {e}")
finally:
    conn.close()

Here's a condensed script that ties everything together for JSON data:

import requests
from zoneinfo import ZoneInfo
from datetime import datetime
import sqlite3

def convert_edt_to_est(edt_str):
    edt_dt = datetime.strptime(edt_str, "%Y-%m-%d %H:%M:%S")
    return edt_dt.replace(tzinfo=ZoneInfo("America/New_York")).astimezone(ZoneInfo("America/New_York"))

def main():
    # Fetch data
    api_url = "https://your-api-endpoint.com/energy-prices"
    try:
        response = requests.get(api_url)
        response.raise_for_status()
        raw_data = response.json()
    except Exception as e:
        print(f"API fetch failed: {e}")
        return

    # Process data
    processed_data = []
    for entry in raw_data:
        try:
            processed_data.append((entry["node_id"], entry["price"], convert_edt_to_est(entry["timestamp"])))
        except KeyError as e:
            print(f"Missing key in entry: {e}")
            continue

    # Insert into SQL
    conn = sqlite3.connect("energy_prices.db")
    cursor = conn.cursor()
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS node_prices (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        node_id TEXT NOT NULL,
        price REAL NOT NULL,
        est_timestamp DATETIME NOT NULL,
        UNIQUE(node_id, est_timestamp)
    )
    """)
    cursor.executemany("INSERT OR IGNORE INTO node_prices VALUES (NULL, ?, ?, ?)", processed_data)
    conn.commit()
    conn.close()
    print(f"Done! Added {cursor.rowcount} valid entries.")

if __name__ == "__main__":
    main()

Quick Tips for New Python Devs: #

• Always add error handling—this keeps your script from crashing when something goes wrong (like a bad API response).
• Test with a small dataset first to verify time conversion works as expected.
• For large datasets (20 years per node!), executemany() is way faster than looping individual execute() calls.

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