集中式Snowflake连接函数在多脚本复用时报错的问题求助

嘿，我来帮你搞定这个问题！你的核心需求是复用Snowflake连接配置，避免在多个脚本里重复修改凭据，还要方便切换DEV/TEST环境，现在遇到的问题其实是两个小细节没处理好，咱们一步步来解决：

先分析你现有代码的核心问题 #

1. 连接函数没有返回可用的连接对象：你的connect_snowflake函数创建了Snowflake连接，但最后没有return conn，就算调用成功，其他脚本也拿不到能执行查询的连接实例。
2. 配置与函数耦合度太高：你把DEV环境的凭据硬编码在脚本的直接调用里，导致其他脚本调用函数时必须重复传参，完全没起到“集中配置”的作用，还触发了TypeError（因为函数要求6个位置参数，你调用时没传）。

解决方案：分两步改造代码 #

第一步：重构你的pySnowflake_Connection.py脚本

我们要把配置逻辑和连接逻辑分离，让函数返回连接对象，同时用环境变量实现“一键切换环境”的需求：

import snowflake.connector
from dotenv import load_dotenv
import os

# 加载环境变量（先安装依赖：pip install python-dotenv）
# 会自动读取当前目录下的.env文件，切换环境只需要改这个文件即可
load_dotenv()

def connect_snowflake():
    conn = None
    try:
        # 从环境变量读取配置，彻底和代码解耦
        conn = snowflake.connector.connect(
            user=os.getenv("SNOWFLAKE_USER"),
            password=os.getenv("SNOWFLAKE_PASSWORD"),
            account=os.getenv("SNOWFLAKE_ACCOUNT"),
            warehouse=os.getenv("SNOWFLAKE_WAREHOUSE"),
            database=os.getenv("SNOWFLAKE_DATABASE"),
            schema=os.getenv("SNOWFLAKE_SCHEMA")
        )
        print("✅ Connection to Snowflake successful!")

        # 验证连接
        cur = conn.cursor()
        cur.execute("SELECT CURRENT_VERSION()")
        version = cur.fetchone()
        print(f" Snowflake Version: {version[0]}")

    except Exception as e:
        print(f"❌ Failed to connect to Snowflake: {e}")
    
    # 关键：返回创建好的连接对象，供其他脚本使用
    return conn

# 只有直接运行这个脚本时，才执行连接测试（方便单独验证）
if __name__ == "__main__":
    test_conn = connect_snowflake()
    if test_conn:
        test_conn.close()

然后在同目录下创建.env文件（用来存环境凭据）：

# DEV环境配置
SNOWFLAKE_USER=USER_DEV
SNOWFLAKE_PASSWORD=xxxxxxxxxx
SNOWFLAKE_ACCOUNT=xxxxxxx.north-europe.azure
SNOWFLAKE_WAREHOUSE=xxxxxxxx_DEV_WH
SNOWFLAKE_DATABASE=xxxxxxxxx_DEV_DB
SNOWFLAKE_SCHEMA=xxxxxxxxxxxx

如果要切换到TEST环境，只需要把.env里的内容改成TEST环境的凭据即可，完全不用修改代码！

第二步：在其他脚本中正确复用连接

现在你的第二个脚本可以改成这样，既不用重复传参，又能拿到可用的连接对象：

import pandas as pd
from pySnowflake_Connection import connect_snowflake

# 获取Snowflake连接对象
connsnf = connect_snowflake()

# 确保连接成功后再执行操作
if connsnf:
    try:
        table_query = "select * from TABLE_DATA_STG"
        # 用pandas的read_sql_query直接读取数据，更简洁
        datafetch = pd.read_sql_query(table_query, connsnf)
        
        # 导出到CSV
        output_path = 'C:/xxxx/Project/PyDataLoad/TABLE_DATA_STG.csv'
        datafetch.to_csv(output_path, sep='|', encoding='utf-8', doublequote=True)
        print(f"✅ 成功导出{len(datafetch)}条数据到 {output_path}！")
    except Exception as e:
        print(f"❌ 查询/导出失败：{e}")
    finally:
        # 用完连接一定要关闭，避免资源泄漏
        connsnf.close()

额外优化建议 #

1. 安全起见：不要把.env文件提交到代码仓库，在.gitignore里加上.env，防止凭据泄露。
2. 多环境快速切换：可以把不同环境的配置分成.env.dev、.env.test，然后通过指定文件名加载：

   # 加载TEST环境配置
   load_dotenv(".env.test")
   

3. 自动管理连接生命周期：用上下文管理器with语句自动关闭连接，避免忘记关闭的问题：

   with connect_snowflake() as conn:
       datafetch = pd.read_sql_query(table_query, conn)
   

备注：内容来源于stack exchange，提问作者Kunal Rane