使用Alembic+SQLAlchemy操作PostgreSQL时数据反复删增问题求助

问题描述 #

使用Alembic结合SQLAlchemy操作PostgreSQL数据库时出现异常：

• 首次运行代码，数据成功写入数据库；
• 再次触发代码，数据库内所有数据被完全删除；
• 第三次运行代码，数据又重新写入。

预期的同步逻辑：

• 若new_data中的记录已存在于数据库，则更新该记录；
• 若new_data中的记录不存在于数据库，则写入该记录；
• 若数据库中的记录不存在于new_data，则删除该记录。

附代码 #

from ConnectDB import create_db_engine
from Models import VirtualServerCustomModel
import ssl

ssl._create_default_https_context = ssl._create_unverified_context

engine, session = create_db_engine()

class VirtualServerCustomService:
    def __init__(self):
        pass
    
    def get_custom_services(self):
        ........

    def db_process(self):
        new_data = self.get_custom_services()

        db_data = session.query(VirtualServerCustomModel).all()

        # 对比新数据中的记录
        for veri in new_data:
            db_verisi_filtreli = session.query(VirtualServerCustomModel).filter_by(vm_name=veri[2]).first()

            if not db_verisi_filtreli:
                # 如果数据库中无此记录，添加新记录
                session.add(VirtualServerCustomModel(*veri))
            
            else:
                # 如果数据库中有此记录，更新字段
                db_verisi_filtreli.vcenter = veri[0]
                db_verisi_filtreli.cluster = veri[1]
                db_verisi_filtreli.host = veri[3]
                db_verisi_filtreli.power_state = veri[4]
                db_verisi_filtreli.vlan = veri[5]
                db_verisi_filtreli.ip_address = veri[6]

        # 删除数据库中有但新数据中没有的记录
        for db_verisi_entry in db_verisi:
            if (db_verisi_entry.vcenter,db_verisi_entry.cluster,db_verisi_entry.host,db_verisi_entry.power_state,db_verisi_entry.vlan, db_verisi_entry.ip_address) not in yeni_veri:
                session.delete(db_verisi_entry)

        session.commit()
        session.close()
        
if __name__ == "__main__":
    VirtualServerCustomService().db_process()

错误排查与修复方案 #

1. 变量名错误（直接触发全量删除） #

• 删除循环中使用了未定义的db_verisi变量，正确的应该是前面查询得到的db_data；
• 判断条件中使用了未定义的yeni_veri变量，正确的应该是new_data。

2. 存在性判断逻辑错误（核心问题） #

当前判断数据库记录是否存在于new_data时，用的是(vcenter, cluster, host, power_state, vlan, ip_address)的字段组合，但new_data中的每个元素是包含vm_name的完整元组，这个组合不可能匹配new_data中的元素，导致所有数据库记录都会被判定为不在新数据中，从而被全部删除。

正确的判断逻辑应该基于唯一标识vm_name实现：

• 先把new_data中的所有vm_name提取出来，存成一个集合；
• 遍历数据库记录时，检查该记录的vm_name是否不在这个集合中，若是则删除。

3. Session管理问题 #

全局创建的session在第一次调用session.close()后就失效了，后续运行会出现会话异常，应该在每次db_process方法内部创建新的session，而非全局复用。

修复后的代码示例 #

from ConnectDB import create_db_engine
from Models import VirtualServerCustomModel
import ssl

ssl._create_default_https_context = ssl._create_unverified_context

class VirtualServerCustomService:
    def __init__(self):
        # 每次实例化时创建新的session
        self.engine, self.session = create_db_engine()
    
    def get_custom_services(self):
        ........

    def db_process(self):
        new_data = self.get_custom_services()
        # 提取新数据中的所有vm_name，用于快速判断
        new_vm_names = {veri[2] for veri in new_data}

        db_data = self.session.query(VirtualServerCustomModel).all()

        # 处理新增/更新逻辑
        for veri in new_data:
            vm_name = veri[2]
            db_record = self.session.query(VirtualServerCustomModel).filter_by(vm_name=vm_name).first()

            if not db_record:
                self.session.add(VirtualServerCustomModel(*veri))
            else:
                db_record.vcenter = veri[0]
                db_record.cluster = veri[1]
                db_record.host = veri[3]
                db_record.power_state = veri[4]
                db_record.vlan = veri[5]
                db_record.ip_address = veri[6]

        # 处理删除逻辑
        for db_entry in db_data:
            if db_entry.vm_name not in new_vm_names:
                self.session.delete(db_entry)

        self.session.commit()
        self.session.close()
        
if __name__ == "__main__":
    VirtualServerCustomService().db_process()

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