SQLAlchemy联合主键重复错误排查：为何不同Date仍触发IntegrityError？

使用SQLAlchemy结合PyMySQL开发医疗预约系统时，遇到如下IntegrityError错误：

(pymysql.err.IntegrityError) (1062, "Duplicate entry '1-6' for key 'exam_schedule.PRIMARY'") [SQL: INSERT INTO exam_schedule (exam_time_id, doctor_id, date, is_book, exam_registration_id) VALUES (%(exam_time_id)s, %(doctor_id)s, %(date)s, %(is_book)s, %(exam_registration_id)s)] [parameters: {'exam_time_id': '1', 'doctor_id': '6', 'date': datetime.date(2024, 11, 26), 'is_book': 1, 'exam_registration_id': 38}]

模型类 #

class ExamSchedule(db.Model):
    exam_time_id = Column(Integer, ForeignKey('exam_time.id'), nullable=False,primary_key=True)
    doctor_id = Column(Integer, ForeignKey('doctor.id'), nullable=False,primary_key=True)
    date = Column(Date, nullable=False,primary_key=True)
    is_book = Column(Boolean, nullable=False, default=True)
    exam_registration_id = Column(Integer, ForeignKey('exam_registration.id', ondelete='CASCADE'), nullable=False,unique=True)

class ExamTime(db.Model):
    id = Column(Integer, primary_key=True)
    start_time = Column(db.Time, nullable=False)
    end_time = Column(db.Time, nullable=False)
    exam_dates = db.relationship('ExamSchedule', backref='exam_time', lazy = True)

class ExamRegistration(db.Model): 
    id = Column(Integer, primary_key=True, )
    symptom = Column(String(255), nullable = True)
    exam_schedule = relationship("ExamSchedule", backref="exam_registration", uselist=False)
    patient_id = Column(Integer, ForeignKey('patient.id'),nullable = False)
    doctor_id = Column(Integer, ForeignKey('doctor.id'),nullable = False)
    is_waiting = Column(Boolean, default=True)
    doctor = relationship("Doctor", backref="exam_registrations")

class User(db.Model):
    id = Column(Integer, primary_key=True)
    last_name = Column(String(50), nullable=False)
    first_name = Column(String(50), nullable=False)
    gender = Column(String(50))
    birth_day = Column(Date)
    email = Column(String(50), unique=True, nullable=False)
    image = Column(String(255), nullable=True)

    phone_numbers = relationship(
        'PhoneNumber',
        backref='user',
        cascade='all, delete-orphan',  
        lazy=True
    )
    
    account_id = Column(Integer, ForeignKey('account.id', ondelete='CASCADE'), nullable=False, unique=True)
    role = Column(String(50), nullable=False, default="user")  # 添加role字段
    __mapper_args__ = {
        'polymorphic_identity': 'user',  # 默认类型
        'polymorphic_on': role          # 基于role区分角色
    }
    

class Doctor(User):
    id = Column(Integer, ForeignKey('user.id', ondelete='CASCADE'), primary_key=True)
    specialty = Column(String(100), nullable=True) 
    degree = Column(String(100), nullable=True)  
    experience = Column(String(100),nullable=True)
    current_workplace = Column(String(100),nullable=True)
    exam_dates = relationship('ExamSchedule', backref='doctor', lazy=True)

    __mapper_args__ = {
        'polymorphic_identity': 'doctor',  
    }

视图与业务逻辑 #

@login_required
def confirm_appoint():
    if request.method == 'POST':
        if not current_user.is_authenticated:
            return redirect(url_for('auth.user_login'))
        else:
            doctor_id = request.form.get('doctor_id')
            exam_time_id =request.form.get('exam_time_id')
            symptom = str(request.form.get('symptom'))
            exam_day = datetime.strptime(request.form.get('exam_day'),  "%Y-%m-%d %H:%M:%S").date()
            if not check_existing_schedule(exam_time_id,doctor_id,exam_day):
                exam_registration = add_apointment(symptom=symptom, doctor_id=doctor_id, patient_id=current_user.user.id)
                add_exam_scheduled(exam_time_id,doctor_id,exam_day, exam_registration.id)
            return redirect(url_for('appointment.appointment_main'))

def add_apointment(symptom, doctor_id, patient_id):
    new_exam_registration = ExamRegistration(
        symptom=symptom,
        doctor_id=doctor_id,
        patient_id=patient_id,
    )
    db.session.add(new_exam_registration)
    db.session.commit()
    return new_exam_registration

def add_exam_scheduled(exam_time_id,doctor_id,date, exam_registration_id):
   
    new_exam_scheduled = ExamSchedule(
        exam_time_id=exam_time_id,
        doctor_id=doctor_id,
        date=date,
        is_book=True,
        exam_registration_id=exam_registration_id
    )
    db.session.add(new_exam_scheduled)
    db.session.commit()
def check_existing_schedule(exam_time_id,doctor_id,date):
    existing_schedule = db.session.query(ExamSchedule).filter_by(exam_time_id=exam_time_id, doctor_id=doctor_id, date=date).first()
    if existing_schedule:
        return True
    else:
        return False

尝试插入两条仅date字段不同的记录：

1. exam_time_id=1, doctor_id=6, date = 2024-11-25
2. exam_time_id=1, doctor_id=6, date = 2024-11-26

明明已将exam_time_id、doctor_id、date设为联合主键，却仍提示主键重复？

从错误提示Duplicate entry '1-6' for key 'exam_schedule.PRIMARY'可以看出，数据库识别的联合主键仅包含exam_time_id和doctor_id两个字段，并未包含date字段。核心原因是数据库表的实际结构与模型定义不一致：

• 可能是最初创建表时，ExamSchedule模型中的date字段未设置primary_key=True，后续修改模型后未同步更新数据库表结构；
• 也可能是使用数据库迁移工具（如Flask-Migrate）时，未生成并执行正确的迁移脚本，导致表结构未更新。

另外，业务逻辑中从表单获取的doctor_id和exam_time_id是字符串类型，而模型定义为Integer，虽然SQLAlchemy通常会自动转换，但也可能引发潜在的类型匹配问题。

1. 同步数据库表结构

   • 先手动检查数据库中exam_schedule表的主键配置，确认是否包含date字段。
   • 如果主键缺少date，执行SQL语句修改表结构（以MySQL为例）：

     ALTER TABLE exam_schedule DROP PRIMARY KEY, ADD PRIMARY KEY (exam_time_id, doctor_id, date);
     

   • 若使用Flask-Migrate，生成新的迁移脚本并执行：

     flask db migrate -m "update exam_schedule primary key"
     flask db upgrade
     

2. 修正业务逻辑中的类型转换
   将表单获取的doctor_id和exam_time_id转换为整数类型，避免类型不匹配：

   # 在confirm_appoint函数中修改
   doctor_id = int(request.form.get('doctor_id'))
   exam_time_id = int(request.form.get('exam_time_id'))
   

3. 优化事务处理
   当前业务逻辑中add_apointment和add_exam_scheduled分别执行db.session.commit()，如果添加ExamSchedule失败，已创建的ExamRegistration不会回滚，可能导致数据冗余。建议将两个操作放在同一个事务中：

   def confirm_appoint():
       if request.method == 'POST':
           if not current_user.is_authenticated:
               return redirect(url_for('auth.user_login'))
           try:
               doctor_id = int(request.form.get('doctor_id'))
               exam_time_id = int(request.form.get('exam_time_id'))
               symptom = str(request.form.get('symptom'))
               exam_day = datetime.strptime(request.form.get('exam_day'),  "%Y-%m-%d %H:%M:%S").date()
               if not check_existing_schedule(exam_time_id,doctor_id,exam_day):
                   # 开启事务，统一提交
                   exam_registration = ExamRegistration(
                       symptom=symptom,
                       doctor_id=doctor_id,
                       patient_id=current_user.user.id,
                   )
                   db.session.add(exam_registration)
                   new_exam_scheduled = ExamSchedule(
                       exam_time_id=exam_time_id,
                       doctor_id=doctor_id,
                       date=exam_day,
                       is_book=True,
                       exam_registration_id=exam_registration.id
                   )
                   db.session.add(new_exam_scheduled)
                   db.session.commit()
               return redirect(url_for('appointment.appointment_main'))
           except Exception as e:
               db.session.rollback()
               # 处理异常，比如返回错误提示
               return redirect(url_for('appointment.appointment_main', error=str(e)))
   

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