如何在Python中依次调用API列表并将数据存入Snowflake

前置准备 #

1. API列表文件：创建api_list.txt，每行填写一个Genesys Cloud的实际API端点（注意你提供的是API文档链接，需提取真实调用地址，例如从文档链接标识#get-api-v2-alerting-alerts-active对应真实端点https://api.genesys.cloud/api/v2/alerting/alerts/active）
2. 依赖安装：执行以下命令安装所需包：

   pip install pandas requests sqlalchemy snowflake-sqlalchemy python-dotenv
   

3. 配置文件：创建.env文件存储敏感配置，避免硬编码：

   # Genesys Cloud OAuth2认证
   GENESYS_CLIENT_ID=你的客户端ID
   GENESYS_CLIENT_SECRET=你的客户端密钥
   GENESYS_REGION=你的Genesys区域（如us-east-1）
   
   # Snowflake连接信息
   SNOWFLAKE_ACCOUNT=你的Snowflake账号
   SNOWFLAKE_USER=你的用户名
   SNOWFLAKE_PASSWORD=你的密码
   SNOWFLAKE_WAREHOUSE=使用的仓库
   SNOWFLAKE_DATABASE=目标数据库
   SNOWFLAKE_SCHEMA=目标Schema
   

核心脚本实现 #

import os
import requests
import pandas as pd
from sqlalchemy import create_engine
from dotenv import load_dotenv
from requests.auth import HTTPBasicAuth

# 加载环境变量
load_dotenv()

def get_genesys_token():
    """获取Genesys Cloud的访问令牌"""
    token_url = f"https://login.{os.getenv('GENESYS_REGION')}.genesys.cloud/oauth/token"
    auth = HTTPBasicAuth(os.getenv('GENESYS_CLIENT_ID'), os.getenv('GENESYS_CLIENT_SECRET'))
    payload = {"grant_type": "client_credentials"}
    response = requests.post(token_url, auth=auth, data=payload)
    response.raise_for_status()
    return response.json()["access_token"]

def get_snowflake_engine():
    """创建Snowflake的SQLAlchemy连接引擎"""
    conn_str = (
        f"snowflake://{os.getenv('SNOWFLAKE_USER')}:{os.getenv('SNOWFLAKE_PASSWORD')}"
        f"@{os.getenv('SNOWFLAKE_ACCOUNT')}/"
        f"{os.getenv('SNOWFLAKE_DATABASE')}/{os.getenv('SNOWFLAKE_SCHEMA')}"
        f"?warehouse={os.getenv('SNOWFLAKE_WAREHOUSE')}"
    )
    return create_engine(conn_str)

def fetch_api_data(api_endpoint, token):
    """调用API并将返回数据转为DataFrame"""
    headers = {"Authorization": f"Bearer {token}"}
    response = requests.get(api_endpoint, headers=headers)
    response.raise_for_status()
    api_data = response.json()
    # 根据API返回结构调整，多数Genesys API数据在'entities'字段中
    return pd.DataFrame(api_data.get('entities', api_data))

def main():
    # 读取API列表
    with open('api_list.txt', 'r') as f:
        api_endpoints = [line.strip() for line in f if line.strip()]
    
    # 获取认证令牌与Snowflake连接
    genesys_token = get_genesys_token()
    snowflake_engine = get_snowflake_engine()
    
    # 遍历处理每个API
    for endpoint in api_endpoints:
        # 自动生成表名（可根据需求自定义）
        table_name = '_'.join(endpoint.split('/')[-2:]).lower()
        print(f"处理API: {endpoint} → 写入表: {table_name}")
        
        # 拉取数据并写入Snowflake
        df = fetch_api_data(endpoint, genesys_token)
        df.to_sql(
            name=table_name,
            con=snowflake_engine,
            if_exists='replace',  # 可选：'append'追加数据，'fail'若表存在则报错
            index=False,
            chunksize=10000  # 大数据量时分批写入，避免内存溢出
        )
        print(f"完成写入：{len(df)} 条数据存入 {table_name}")

if __name__ == "__main__":
    main()

关键注意事项 #

• API认证：确保Genesys的客户端ID/密钥拥有对应API的访问权限，若权限不足会返回403错误
• 数据结构适配：不同Genesys API的返回结构可能不同，需调整fetch_api_data中的数据提取逻辑（比如部分API可能返回results而非entities）
• 分页处理：如果API返回的数据量超过单页上限，需添加分页逻辑（检查返回结果中的nextUri字段，循环请求直到获取所有数据）
• 表名与写入策略：table_name的生成规则可根据业务需求修改，if_exists参数需根据是覆盖还是追加数据选择对应值

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