在类__init__中使用sqlalchemy.create_engine()触发ProgrammingError问题

报错提示找不到表对象，核心原因是类中初始化的Engine在复用过程中，连接池内的连接状态异常：

• 函数实现里每次调用都会新建Engine，连接是全新的，能正确解析数据库对象；
• 类实现里Engine在初始化时创建，后续调用方法复用该Engine时，连接池内的连接可能因闲置超时、状态异常等问题，导致数据库无法正确识别三部分命名的表路径。

方案1：显式管理连接生命周期 #

在类的方法中使用上下文管理器获取活跃连接，确保每次查询都使用有效连接：

class SqlDownload():
    
    def __init__(self):
        driver="ODBC Driver 17 for SQL Server"
        # 注意：实际使用时请补充完整的连接字符串（协议、服务器、用户名密码等）
        connection_url = f"mssql+pyodbc://username:password@server/db_Dev?driver={driver.replace(' ', '+')}"
        self.engine = sa.create_engine(connection_url)
    
    def get_uploaded(self):
        # 显式获取并管理连接
        with self.engine.connect() as conn:
            query = "SELECT DISTINCT name FROM [db_Dev].[de].[ti]"
            df = pd.read_sql(sql=query, con=conn)
            uploaded_list = df["name"].to_list()
            return uploaded_list

# 调用测试
d = SqlDownload()
d.get_uploaded()

方案2：延迟初始化Engine #

将Engine的创建延迟到第一次调用方法时，避免提前创建的Engine因闲置出现状态问题：

class SqlDownload():
    
    def __init__(self):
        self.driver="ODBC Driver 17 for SQL Server"
        self.connection_url = f"mssql+pyodbc://username:password@server/db_Dev?driver={self.driver.replace(' ', '+')}"
        self.engine = None
    
    def get_uploaded(self):
        if not self.engine:
            self.engine = sa.create_engine(self.connection_url)
        query = "SELECT DISTINCT name FROM [db_Dev].[de].[ti]"
        df = pd.read_sql(sql=query, con=self.engine)
        uploaded_list = df["name"].to_list()
        return uploaded_list

# 调用测试
d = SqlDownload()
d.get_uploaded()

额外注意事项 #

• 连接字符串必须完整：确保包含mssql+pyodbc://协议头、服务器地址、认证信息、数据库名，且Driver名称中的空格要替换为+（URL编码要求）；
• 复用Engine时，可通过SQLAlchemy的连接池配置（如pool_recycle参数）避免连接超时：

  self.engine = sa.create_engine(connection_url, pool_recycle=3600)
  

  该参数会自动回收闲置超过3600秒的连接，避免状态异常。

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