如何在多个Python类中实现可复用数据库插入逻辑，避免代码重复

核心思路 #

结合模板方法模式和策略模式，把插入操作的固定流程和可变逻辑完全分离：

• 固定流程（创建连接、提交/回滚事务、关闭连接）封装在抽象基类的模板方法中，仅需编写1次
• 每个插入场景的可变部分（查询语句、参数构造、错误处理）封装为独立的具体策略类，继承抽象基类实现对应抽象方法
• 业务类按需组合多个插入策略实例，即可实现同个类下多套不同插入逻辑的需求，完全避免重复代码

具体实现 #

第一步：定义抽象插入策略基类 #

基类实现固定的插入流程模板，同时声明3个可变点的抽象方法：

from abc import ABC, abstractmethod
# 假设已有open_connection、close_connection工具方法，且IntegrityError已导入
class BaseInsertStrategy(ABC):
    def insert(self):
        # 固定流程完全复用，无需重复编写
        db, cursor = open_connection()
        try:
            query = self._get_query()
            params = self._get_params()
            # 支持批量执行，单条/批量逻辑都可以在_do_execute里实现
            self._do_execute(cursor, query, params)
            db.commit()
        except IntegrityError as e:
            self._handle_error(e, db)
        finally:
            close_connection(db, cursor)
    
    @abstractmethod
    def _get_query(self):
        """返回当前插入的SQL语句，子类必须实现"""
        pass
    
    @abstractmethod
    def _get_params(self):
        """返回当前插入的参数（单条参数/批量参数列表都可），子类必须实现"""
        pass
    
    @abstractmethod
    def _handle_error(self, error, db_conn):
        """当前插入的专属错误处理逻辑，子类必须实现"""
        pass
    
    def _do_execute(self, cursor, query, params):
        """默认执行逻辑，批量插入的子类可以重写该方法"""
        cursor.execute(query, params)

第二步：为每个插入场景实现具体策略类 #

每个插入逻辑单独封装为一个类，仅需实现3个可变方法即可，无需再编写重复的流程代码：

# 插入model的专属策略
class InsertModelStrategy(BaseInsertStrategy):
    def __init__(self, model_name):
        self.model_name = model_name
    
    def _get_query(self):
        return "INSERT INTO models(name) VALUES (%s);"
    
    def _get_params(self):
        return (self.model_name,)
    
    def _handle_error(self, error, db_conn):
        print(f"model.name 已存在，值为：{self.model_name}")
        db_conn.rollback()

# 插入objects的专属策略
class InsertObjectsStrategy(BaseInsertStrategy):
    def __init__(self, model_name, objects):
        self.model_name = model_name
        self.objects = objects
    
    def _get_query(self):
        return """INSERT INTO objects (name, model_id)
        VALUES (%s,(SELECT id FROM models WHERE name=%s));"""
    
    def _get_params(self):
        # 批量参数构造
        return [(obj, self.model_name) for obj in self.objects]
    
    def _do_execute(self, cursor, query, params):
        # 重写执行逻辑为批量循环执行
        for param in params:
            cursor.execute(query, param)
    
    def _handle_error(self, error, db_conn):
        if error.msg == "Column 'model_id' cannot be null":
            print("model_id 不存在，请先插入对应的model")
        if error.msg == "Duplicate entry 'perrito-3' for key 'name'":
            print(f"model.name-object.name 已存在，对应model：{self.model_name}")
        db_conn.rollback()

# 插入user的专属策略
class InsertUserStrategy(BaseInsertStrategy):
    def __init__(self, user_id, user_name):
        self.user_id = user_id
        self.user_name = user_name
    
    def _get_query(self):
        return "INSERT INTO users (id, name) VALUES (%s,%s);"
    
    def _get_params(self):
        return (self.user_id, self.user_name)
    
    def _handle_error(self, error, db_conn):
        print(f"user.id 已存在，值为：{self.user_id}")
        db_conn.rollback()

第三步：业务类按需调用策略即可 #

业务类不需要继承任何类，需要多少种插入逻辑，就实例化多少种对应的策略类调用即可：

class One:
    def __init__(self, name):
        self.name = name
    
    def insert_objects(self, objects=None):
        # 调用插入objects的策略
        InsertObjectsStrategy(self.name, objects).insert()
    
    def insert_model(self):
        # 调用插入model的策略
        InsertModelStrategy(self.name).insert()

class Two:
    def __init__(self, user_id, user_name):
        self.id = user_id
        self.name = user_name
    
    def insert_user(self):
        # 调用插入user的策略
        InsertUserStrategy(self.id, self.name).insert()

方案优势 #

• 完全消除重复代码：固定插入流程仅写一次，所有插入场景复用
• 完美适配多插入实现需求：每个插入逻辑独立封装，同个业务类可以引入任意多个插入策略，互不干扰
• 通用OOP实现：完全基于类继承和组合，没有用到Python特有的可调用对象传参，所有面向对象语言都可以直接复用该方案
• 扩展性极强：新增插入场景只需要新增一个策略类实现3个抽象方法，不需要修改原有代码，符合开闭原则

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