Flask+MySQL中SQLAlchemy bulk_save_objects批量插入datetime无返回值问题

问题背景 #

在Flask+MySQL环境下，使用bulk_save_objects批量插入大量学生成绩记录以优化性能，但插入后返回的StudentScore对象中，created_at和updated_at字段始终为null。这两个字段在模型中已通过server_default=db.func.now()设置数据库默认值，但无法获取到数据库生成的实际值。考虑过使用add_all()但担心性能不足。

模型代码 #

from database.shared_db import db
from entities.student_score import StudentScore


class StudentScoresModel(db.Model):
    __tablename__ = 'students_scores'

    id = db.Column(db.VARCHAR(36), primary_key=True)
    student_id = db.Column(db.VARCHAR(36), db.ForeignKey('students.id'), nullable=False)
    attribute_id = db.Column(db.VARCHAR(36), db.ForeignKey('attributes.id'), nullable=False)
    score = db.Column(db.Float, nullable=False)
    updated_at = db.Column(
        db.DateTime, 
        server_default=db.func.now(), 
        onupdate=db.func.now(), 
        nullable=False,
    )
    created_at = db.Column(
        db.DateTime, 
        server_default=db.func.now(), 
        nullable=False,
    )

    def convert_to_entity(self) -> StudentScore:
        return StudentScore(
            id=self.id,
            student_id=self.student_id,
            attribute_id=self.attribute_id,
            score=self.score,
            updated_at=self.updated_at,
            created_at=self.created_at,
        )

批量插入代码 #

@staticmethod
def create_student_scores(
    student_scores: list[StudentScore],
) -> list[StudentScore]:
    student_scores_model: list[StudentScoresModel] = [
        StudentScoresModel(
            id=str(uuid.uuid4()),                                                                             
            attribute_id=student_Score.attribute_id,                                                                             
            score=student_Score.score,                                                                             
            student_id=student_Score.student_id
        ) for student_Score in student_scores
    ]
    db.session.bulk_save_objects(
        student_scores_model, 
        return_defaults=True,
    )
    db.session.commit()
    return [
       student_score.convert_to_entity() 
       for student_score in student_scores_model
    ]

问题原因 #

bulk_save_objects的return_defaults=True仅支持自增主键这类数据库自动生成的主键值，对于server_default定义的非主键字段（如datetime类型的时间字段），批量操作不会主动将数据库生成的值同步回ORM对象——这是因为批量操作为了性能，跳过了ORM对象的完整生命周期管理，也不会自动执行额外查询去获取这些非主键的默认值。

解决办法 #

方法1：批量查询刷新对象（需立即获取字段值） #

提交事务后，通过主键批量查询刚插入的记录，将数据库生成的时间字段同步回对象：

@staticmethod
def create_student_scores(
    student_scores: list[StudentScore],
) -> list[StudentScore]:
    student_scores_model: list[StudentScoresModel] = [
        StudentScoresModel(
            id=str(uuid.uuid4()),                                                                             
            attribute_id=student_Score.attribute_id,                                                                             
            score=student_Score.score,                                                                             
            student_id=student_Score.student_id
        ) for student_Score in student_scores
    ]
    # 执行批量插入
    db.session.bulk_save_objects(student_scores_model, return_defaults=True)
    db.session.commit()
    
    # 批量查询获取包含时间字段的最新记录
    inserted_ids = [model.id for model in student_scores_model]
    refreshed_models = StudentScoresModel.query.filter(StudentScoresModel.id.in_(inserted_ids)).all()
    
    # 转换为实体类返回
    return [model.convert_to_entity() for model in refreshed_models]

该方法会多一次批量查询，性能略低于纯批量插入，但远优于add_all()，适合必须返回完整字段的场景。

方法2：代码中生成时间（性能最优） #

如果业务允许不依赖数据库服务器时间，直接在Python代码中生成created_at和updated_at的值，完全避免数据库往返：

from datetime import datetime

@staticmethod
def create_student_scores(
    student_scores: list[StudentScore],
) -> list[StudentScore]:
    current_time = datetime.now()
    student_scores_model: list[StudentScoresModel] = [
        StudentScoresModel(
            id=str(uuid.uuid4()),                                                                             
            attribute_id=student_Score.attribute_id,                                                                             
            score=student_Score.score,                                                                             
            student_id=student_Score.student_id,
            created_at=current_time,
            updated_at=current_time
        ) for student_Score in student_scores
    ]
    db.session.bulk_save_objects(student_scores_model)
    db.session.commit()
    return [model.convert_to_entity() for model in student_scores_model]

这种方法性能最高，逻辑可控，适合对时间精度要求不严格依赖数据库的场景。

方法3：改用add_all()（小数据量场景） #

如果插入数据量不大（如几千条以内），add_all()虽然性能不如批量操作，但会完整触发ORM生命周期，提交后对象会自动获取数据库生成的默认值：

@staticmethod
def create_student_scores(
    student_scores: list[StudentScore],
) -> list[StudentScore]:
    student_scores_model: list[StudentScoresModel] = [
        StudentScoresModel(
            id=str(uuid.uuid4()),                                                                             
            attribute_id=student_Score.attribute_id,                                                                             
            score=student_Score.score,                                                                             
            student_id=student_Score.student_id
        ) for student_Score in student_scores
    ]
    db.session.add_all(student_scores_model)
    db.session.commit()
    return [model.convert_to_entity() for model in student_scores_model]

注意：数据量上万级时，add_all()会因ORM对象开销导致性能明显下降，不推荐使用。

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