将Jupyter Notebook天气数据API获取及DynamoDB写入代码转换为Python函数时的故障排查咨询

首先，你的代码在Jupyter中能运行但封装成函数失败，核心问题大概率是全局变量残留、关键变量赋值错误和缺乏异常调试信息导致的。我帮你重构代码并添加完整的调试逻辑，同时修复几个致命问题：

核心错误修复与调试点 #

1. 致命错误：api_key赋值错误
   你写了api_key = print("\"%s\""% ssmkey )，但print()函数的返回值是None！Jupyter中可能因为之前运行过正确的api_key赋值（变量残留）所以能正常调用API，但封装成函数后，api_key变成None，直接导致API请求失败。

2. 资源初始化的作用域问题
   把boto3客户端、geolocator等资源初始化放在函数内部，避免全局变量在函数调用时的上下文冲突。

3. 添加调试打印与异常捕获
   在关键步骤（获取SSM参数、API请求、DynamoDB写入）添加打印，同时捕获异常并输出详细错误信息，方便定位问题。

重构后的完整代码 #

import boto3
import pyowm
import time
import json
import requests
from datetime import datetime, date, timedelta, timezone
import pandas as pd
from geopy.geocoders import Nominatim
from geopy.exc import GeocoderTimedOut, GeocoderUnavailable

def fetch_and_save_weather(city, table_name):
    # 调试：打印函数启动信息
    print(f"=== 开始处理城市：{city} ===")
    
    try:
        # 1. 获取SSM中的API密钥
        print("步骤1：获取OpenWeather API密钥...")
        client = boto3.client('ssm')
        parameter = client.get_parameter(Name='OpenWeather_API_Key', WithDecryption=False)
        ssmkey = parameter['Parameter']['Value']
        api_key = ssmkey  # 修复：直接赋值，不要用print
        print(f"成功获取API密钥：{api_key[:4]}...（隐藏部分内容）")

        # 2. 地理编码获取经纬度
        print("步骤2：获取城市经纬度...")
        geolocator = Nominatim(user_agent='aerieous@myserver.com')
        location = geolocator.geocode(city, timeout=10)
        if not location:
            raise ValueError(f"无法找到城市 {city} 的地理信息")
        lat = location.latitude
        lon = location.longitude
        print(f"城市 {city} 的经纬度：纬度 {lat:.4f}，经度 {lon:.4f}")

        # 3. 计算昨日时间戳（epoch格式）
        print("步骤3：计算昨日时间戳...")
        x = (datetime.now() - timedelta(days=1))
        d = x.isoformat(' ', 'seconds')
        p = '%Y-%m-%d %H:%M:%S'
        dt = int(time.mktime(time.strptime(d, p)))
        print(f"昨日时间戳（epoch）：{dt}")

        # 4. 调用OpenWeather API获取历史数据
        print("步骤4：调用OpenWeather API...")
        url = "https://api.openweathermap.org/data/2.5/onecall/timemachine?lat=%s&lon=%s&dt=%s&appid=%s&units=metric" % (
            lat, lon, dt, api_key)
        response = requests.get(url, timeout=15)
        # 调试：打印响应状态码和头部信息
        print(f"API响应状态码：{response.status_code}")
        if response.status_code != 200:
            raise Exception(f"API请求失败，响应内容：{response.text}")
        data_history = json.loads(response.text)
        print(f"成功获取 {len(data_history.get('hourly', []))} 条小时级天气数据")

        # 5. 处理数据并转换为DataFrame
        print("步骤5：处理天气数据...")
        df_history2 = pd.json_normalize(data_history, record_path='hourly', meta=['lat', 'lon', 'timezone'], errors='ignore')
        df_history2['dt'] = pd.to_datetime(df_history2['dt'], unit='s').dt.strftime("%m/%d/%Y %H:%M:%S")
        df_history2 = df_history2.rename(columns={'dt': 'timestamp'})
        df_history2['uuid'] = df_history2[['timestamp', 'timezone']].agg('-'.join, axis=1)

        df_select_hist2 = df_history2[
            ['uuid', 'lat', 'lon', 'timezone', 'timestamp', 'temp', 'feels_like', 'humidity', 'pressure']]
        df_select_hist2 = df_select_hist2.astype(str)
        content = df_select_hist2.to_dict('records')
        print(f"处理完成，共准备 {len(content)} 条待写入DynamoDB的数据")

        # 6. 写入DynamoDB
        print("步骤6：写入DynamoDB...")
        dynamodb = boto3.resource('dynamodb')
        table = dynamodb.Table(table_name)
        success_count = 0
        for idx, item in enumerate(content):
            try:
                table.put_item(
                    Item={
                        'pk_id': item['uuid'],
                        'sk': item['timestamp'],
                        'gsi_1_pk': item['lat'],
                        'gsi_1_sk': item['lon'],
                        'gsi_2_pk': item['temp'],
                        'gsi_2_sk': item['feels_like'],
                        'humidity': item['humidity'],
                        'pressure': item['pressure'],
                        'timezone': item['timezone']
                    }
                )
                success_count += 1
                # 每5条打印一次进度
                if (idx + 1) % 5 == 0:
                    print(f"已写入 {idx+1}/{len(content)} 条数据")
            except Exception as e:
                print(f"写入第 {idx+1} 条数据失败：{str(e)}")
        
        print(f"=== 处理完成！成功写入 {success_count}/{len(content)} 条数据 ===")
        return success_count

    except GeocoderTimedOut:
        print(f"错误：地理编码请求超时，请稍后重试")
        return 0
    except GeocoderUnavailable:
        print(f"错误：地理编码服务不可用")
        return 0
    except requests.exceptions.RequestException as e:
        print(f"错误：网络请求失败：{str(e)}")
        return 0
    except boto3.exceptions.Boto3Error as e:
        print(f"错误：AWS服务调用失败：{str(e)}")
        return 0
    except Exception as e:
        print(f"未知错误：{str(e)}")
        return 0

# 函数调用测试
if __name__ == "__main__":
    fetch_and_save_weather('Berlin, DE', 'Dev_Weather')

调试与验证步骤 #

1. 运行代码后，查看控制台的打印信息，确认每一步是否正常执行。
2. 如果某一步失败，控制台会输出具体的错误信息，比如API响应状态码、DynamoDB写入错误等，根据提示定位问题。
3. 可以单独测试某一部分逻辑（比如地理编码、API请求），缩小排查范围。

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