Python Sas_session类：检测SAS连接存活及实例访问触发的魔法方法

一、连接存活检测与自动重连 #

利用saspy自带的连接状态检测方法，结合重连逻辑实现自动恢复：

1. 添加连接状态检测方法 #

新增私有方法判断当前连接是否存活，兼容session未初始化或调用异常的情况：

def _is_connection_alive(self):
    if hasattr(self, 'session'):
        try:
            return self.session.isConnected()
        except Exception:
            # 调用检测方法出错也视为连接断开
            return False
    return False

2. 优化重连逻辑 #

修改conectar()方法，先清理旧连接再创建新连接，避免资源残留：

def conectar(self):
    # 尝试关闭旧连接（如果存在）
    if hasattr(self, 'session'):
        try:
            self.session.disconnect()
        except Exception:
            pass
    # 初始化新连接
    self.session = saspy.SASsession(
        cfgname=self.cfgname,
        cfgfile=self.cfgfile,
        omruser=self.omruser,
        iomhost=self.iomhost,
        iomport=self.iomport,
        appserver=self.appserver,
        authkey=self.authkey,
        timeout=self.timeout
    )
    print('conexão instanciada')
    return self

3. 自动重连触发方法 #

封装公共方法，用于检测连接并自动重连：

def ensure_connection(self):
    if not self._is_connection_alive():
        print('连接已断开，正在重连...')
        self.conectar()
    return self.session

二、每次访问实例时执行的魔法方法 #

要实现每次访问实例属性/方法时自动检查连接，可以重写__getattribute__魔法方法。注意需排除基础配置属性，避免递归调用：

def __getattribute__(self, name):
    # 排除配置类属性，避免不必要的连接检查
    exclude_attrs = {'iomhost', 'iomport', 'appserver', 'omruser', 'authkey', 'cfgname', 'timeout', 'cfgfile', '_is_connection_alive', 'conectar', 'ensure_connection'}
    if name not in exclude_attrs:
        # 访问目标属性前确保连接存活
        self.ensure_connection()
    # 调用父类方法返回原属性
    return super().__getattribute__(name)

完整修改后的代码 #

class Sas_session:
    def __init__(self) -> None:
        self.iomhost: str = sasanl.host  # type: ignore
        self.iomport: int = sasanl.port  # type: ignore
        self.appserver: str = sasanl.appserver  # type: ignore
        self.omruser: str = ""  # type: ignore
        self.authkey: str = ""  # type: ignore
        self.cfgname: str = "iomwin"  # type: ignore
        self.timeout: int = 30  # type: ignore
        self.cfgfile = str(Path(__file__).parent / "ConexaoSAS.py")
        self.conectar()
          
    def conectar(self):
        if hasattr(self, 'session'):
            try:
                self.session.disconnect()
            except Exception:
                pass
        self.session = saspy.SASsession(
            cfgname=self.cfgname,
            cfgfile=self.cfgfile,
            omruser=self.omruser,
            iomhost=self.iomhost,
            iomport=self.iomport,
            appserver=self.appserver,
            authkey=self.authkey,
            timeout=self.timeout
        )
        print('conexão instanciada')
        return self
    
    def _is_connection_alive(self):
        if hasattr(self, 'session'):
            try:
                return self.session.isConnected()
            except Exception:
                return False
        return False
    
    def ensure_connection(self):
        if not self._is_connection_alive():
            print('连接已断开，正在重连...')
            self.conectar()
        return self.session
    
    def __getattribute__(self, name):
        exclude_attrs = {'iomhost', 'iomport', 'appserver', 'omruser', 'authkey', 'cfgname', 'timeout', 'cfgfile', '_is_connection_alive', 'conectar', 'ensure_connection'}
        if name not in exclude_attrs:
            self.ensure_connection()
        return super().__getattribute__(name)

关键说明 #

• _is_connection_alive()：通过saspy原生方法检测连接，捕获异常避免崩溃
• ensure_connection()：主动触发连接检查，可单独调用也可被魔法方法自动触发
• __getattribute__：拦截所有属性访问，自动执行连接检查，实现“无感”重连

内容的提问来源于stack exchange，提问作者FábioRB