Cursor技能封装实战：从原理到最佳实践

1次阅读

没有评论

共计 1683 个字符，预计需要花费 5 分钟才能阅读完成。

在开发涉及数据库交互的应用程序时，直接使用 Cursor 对象进行查询和操作是一种常见做法。但随着业务逻辑变得复杂，这种方式会暴露出几个明显问题：

代码重复：相似的 Cursor 操作代码散落在各处，难以维护
状态管理混乱：Cursor 的状态（打开 / 关闭）、异常处理和资源释放容易遗漏
耦合度高：业务逻辑与数据库操作紧密耦合，难以单独测试或替换
错误处理不统一：不同开发者对异常的处理方式可能不一致

直接调用
优点：实现简单，适合快速原型开发
缺点：所有问题都会暴露，维护成本高
简单封装
优点：提取了常用操作，减少重复代码
缺点：功能有限，无法应对复杂场景
高级封装
优点：统一的接口、状态管理和错误处理
缺点：需要更多前期设计工作

基于面向对象思想，我们可以设计一个 SafeCursor 类，它应该具备：

自动资源管理（实现 __enter__ 和__exit__）
统一的错误处理机制
常用查询方法的封装
状态检查

class SafeCursor:
    """安全的 Cursor 封装类，自动处理资源管理和错误"""
    def __init__(self, connection):
        self._connection = connection
        self._cursor = None

    def __enter__(self):
        self._cursor = self._connection.cursor()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self._cursor:
            self._cursor.close()

    def execute_safe(self, query, params=None):
        """
        执行 SQL 查询并自动处理异常
        :param query: SQL 语句
        :param params: 参数
        :return: 受影响的行数
        """
        try:
            if params:
                return self._cursor.execute(query, params)
            else:
                return self._cursor.execute(query)
        except Exception as e:
            # 记录日志等操作
            raise

    def fetch_all(self):
        """获取所有结果"""
        return self._cursor.fetchall()

    def fetch_one(self):
        """获取单条结果"""
        return self._cursor.fetchone()

    @property
    def rowcount(self):
        """获取受影响的行数"""
        return self._cursor.rowcount