如何在多Python脚本/Notebook中复用Snowflake连接无需重复登录

需求 #

在多个Python脚本或Jupyter Notebook中全局复用Snowflake连接（Snowflake支持异步查询），避免重复输入凭证，同时避免在同一Python服务器实例中建立多个连接。

获取用户凭证代码 #

from getpass import getpass
account = input("accountName: ")
user = input("username: ")
password = getpass("password: ")
myCredentials = dict(account=account, user=user, password=password)

尝试过的连接方式 #

使用SQLAlchemy

from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL
engine = create_engine(URL(**myCredentials))
conn_alchemy = engine.connect()

使用Snowflake Connector

import snowflake.connector as snflk
conn_snflk = snflk.connect(**myCredentials)

遇到的问题 #

连接能够成功创建，但无法共享到其他Notebook中。例如使用Jupyter的%store魔法函数时，连接器对象无法被序列化，导致无法跨Notebook复用。

期望实现：在其他脚本/Notebook中直接调用conn_alchemy或conn_snflk执行查询，无需重新登录，且不会出现NameError: name 'conn_alchemy' is not defined错误，比如执行以下代码：

import pandas as pd
query = "SELECT * FROM myTable"
df = pd.read_sql_query(query, conn_alchemy)

可行解决方案 #

方案1：单例模块封装连接（推荐通用场景）

创建独立的Python模块（如snowflake_conn.py），用单例模式封装连接逻辑，确保全局仅初始化一次凭证和连接：

# snowflake_conn.py
from getpass import getpass
from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL
import snowflake.connector as snflk

# 全局变量存储凭证和连接，仅首次初始化
_credentials = None
_alchemy_engine = None
_snflk_conn = None

def get_credentials():
    global _credentials
    if _credentials is None:
        account = input("accountName: ")
        user = input("username: ")
        password = getpass("password: ")
        _credentials = dict(account=account, user=user, password=password)
    return _credentials

def get_alchemy_conn():
    global _alchemy_engine
    if _alchemy_engine is None:
        creds = get_credentials()
        _alchemy_engine = create_engine(URL(**creds))
    return _alchemy_engine.connect()

def get_snflk_conn():
    global _snflk_conn
    if _snflk_conn is None:
        creds = get_credentials()
        _snflk_conn = snflk.connect(**creds)
    return _snflk_conn

在其他脚本/Notebook中导入模块直接使用：

from snowflake_conn import get_alchemy_conn
import pandas as pd

conn_alchemy = get_alchemy_conn()
query = "SELECT * FROM myTable"
df = pd.read_sql_query(query, conn_alchemy)

首次调用时会要求输入凭证，后续调用自动复用已创建的连接，无需重复登录。

方案2：同Jupyter内核变量共享（仅适用于同内核Notebook）

如果多个Notebook共用一个Jupyter内核，可在初始化连接的Notebook中将连接存入全局命名空间：

# 初始化连接的Notebook
from getpass import getpass
from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL

account = input("accountName: ")
user = input("username: ")
password = getpass("password: ")
myCredentials = dict(account=account, user=user, password=password)

engine = create_engine(URL(**myCredentials))
conn_alchemy = engine.connect()

# 将连接写入全局命名空间
globals()['conn_alchemy'] = conn_alchemy

在同内核的其他Notebook中直接调用：

import pandas as pd
query = "SELECT * FROM myTable"
df = pd.read_sql_query(query, conn_alchemy)

注意：关闭内核后连接会失效，需重新初始化。

方案3：连接池复用连接（适合高并发场景）

Snowflake Connector和SQLAlchemy都支持连接池，配置后可自动复用连接，避免重复创建：
对于SQLAlchemy，创建引擎时指定连接池参数：

from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL

# 配置连接池，pool_size为常驻连接数，max_overflow为额外可创建的连接数
engine = create_engine(URL(**myCredentials), pool_size=5, max_overflow=10)
# 每次调用engine.connect()都会从连接池获取可用连接
conn_alchemy = engine.connect()

将引擎创建逻辑封装到模块中，其他脚本导入引擎即可复用连接池中的连接。

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