如何设计无重复连接代码的Python/SQLite应用？

针对你当前的代码重复、数据库耦合问题，以下是符合DRY原则且易于扩展的解决方案：

1. 封装数据库操作工具类（消除重复连接逻辑） #

创建统一的数据库操作抽象层，把连接创建、SQL执行、事务处理等重复逻辑封装起来，所有业务类共用这个工具类，彻底避免重复写连接代码。

import sqlite3
from abc import ABC, abstractmethod

# 数据库操作抽象基类，定义统一接口
class BaseDatabaseHandler(ABC):
    @abstractmethod
    def connect(self):
        pass

    @abstractmethod
    def execute(self, query, params=None):
        pass

# SQLite具体实现类
class SQLiteHandler(BaseDatabaseHandler):
    def __init__(self, db_path):
        self.db_path = db_path
        self.conn = None

    def connect(self):
        # 保持长连接，避免重复创建连接
        if not self.conn:
            self.conn = sqlite3.connect(self.db_path)
        return self.conn

    def execute(self, query, params=None):
        conn = self.connect()
        cur = conn.cursor()
        try:
            if params:
                cur.execute(query, params)
            else:
                cur.execute(query)
            conn.commit()
            # 返回查询结果（如果有）
            return cur.fetchall() if cur.description else None
        except Exception as e:
            conn.rollback()
            raise e
        finally:
            # 可根据需求选择是否关闭游标/连接，SQLite默认连接管理更灵活
            cur.close()

# 全局实例化，所有业务类共享同一个数据库连接
db_handler = SQLiteHandler("mydb.db")

2. 抽象业务对象基类（复用CRUD逻辑） #

创建业务对象基类，封装通用的保存、删除、更新方法。具体业务类只需提供表名、字段映射等专属信息，无需重复编写CRUD代码。

class BaseBusinessObject:
    def __init__(self, db_handler):
        self.db_handler = db_handler
        # 子类必须重写以下属性
        self.table_name = ""
        self.fields = []  # 例如 ["id", "name", "status"]

    def save_to_db(self):
        # 根据id判断是新增还是更新
        if hasattr(self, "id") and self.id is not None:
            # 更新逻辑
            set_clause = ", ".join([f"{field} = ?" for field in self.fields if field != "id"])
            query = f"UPDATE {self.table_name} SET {set_clause} WHERE id = ?"
            params = [getattr(self, field) for field in self.fields if field != "id"] + [self.id]
        else:
            # 新增逻辑
            placeholders = ", ".join(["?" for _ in self.fields])
            query = f"INSERT INTO {self.table_name} ({', '.join(self.fields)}) VALUES ({placeholders})"
            params = [getattr(self, field) for field in self.fields]
        
        self.db_handler.execute(query, params)
        # 新增时自动获取自增id
        if not hasattr(self, "id") or self.id is None:
            self.id = self.db_handler.execute("SELECT last_insert_rowid()")[0][0]

    def delete_from_db(self):
        if not hasattr(self, "id") or self.id is None:
            raise ValueError("无法删除未设置id的对象")
        query = f"DELETE FROM {self.table_name} WHERE id = ?"
        self.db_handler.execute(query, (self.id,))

# 具体业务类示例
class BusinessObject1(BaseBusinessObject):
    def __init__(self, db_handler, id=None, name=None, value=None):
        super().__init__(db_handler)
        self.table_name = "business_object1"
        self.fields = ["id", "name", "value"]
        self.id = id
        self.name = name
        self.value = value

class BusinessObject2(BaseBusinessObject):
    def __init__(self, db_handler, id=None, code=None, description=None):
        super().__init__(db_handler)
        self.table_name = "business_object2"
        self.fields = ["id", "code", "description"]
        self.id = id
        self.code = code
        self.description = description

3. 支持多数据库切换（解耦数据库依赖） #

后续需要切换到MySQL或PostgreSQL时，只需新增对应数据库的BaseDatabaseHandler实现类，替换全局的db_handler实例即可，业务代码完全无需修改。

示例MySQL实现框架：

import mysql.connector

class MySQLHandler(BaseDatabaseHandler):
    def __init__(self, host, user, password, db_name):
        self.host = host
        self.user = user
        self.password = password
        self.db_name = db_name
        self.conn = None

    def connect(self):
        if not self.conn:
            self.conn = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.db_name
            )
        return self.conn

    def execute(self, query, params=None):
        conn = self.connect()
        cur = conn.cursor()
        try:
            if params:
                cur.execute(query, params)
            else:
                cur.execute(query)
            conn.commit()
            return cur.fetchall() if cur.description else None
        except Exception as e:
            conn.rollback()
            raise e
        finally:
            cur.close()

# 切换数据库时替换实例即可
# db_handler = MySQLHandler("localhost", "root", "password", "mydb")

4. 统一维护全局对象信息表 #

在BaseBusinessObject中添加全局日志逻辑，每次保存/删除对象时，自动同步更新全局对象信息表：

class BaseBusinessObject:
    # 原有代码...

    def save_to_db(self):
        # 原有保存逻辑...
        self._update_global_object_log("save")

    def delete_from_db(self):
        # 原有删除逻辑...
        self._update_global_object_log("delete")

    def _update_global_object_log(self, operation):
        from datetime import datetime
        # 假设全局表名为object_log，字段：id, object_type, object_id, operation_time, operation
        query = """
            INSERT INTO object_log (object_type, object_id, operation_time, operation)
            VALUES (?, ?, ?, ?)
        """
        params = (self.table_name, self.id, datetime.now().isoformat(), operation)
        self.db_handler.execute(query, params)

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