如何在PynamoDB批量操作中维护created_at与last_modified_at属性

问题1：确保created_at仅在新增条目时赋值一次 #

你的当前代码中，batch.save()会执行PutItem操作直接覆盖整个条目，导致已有条目的created_at被更新为新时间。同时模型定义里default_for_new=datetime.now(timezone.utc)存在问题——该表达式在类加载时就会执行，所有新实例会共享同一个时间戳，正确写法是传入可调用的lambda函数来动态生成当前时间。

解决方案：使用batch.update_item()结合更新表达式与条件表达式，仅在条目不存在时设置created_at：

1. 先修正模型的时间字段默认值：

from datetime import datetime, timezone
from pynamodb.models import Model
from pynamodb.attributes import UnicodeAttribute, MapAttribute, UTCDateTimeAttribute

class ArticleModel(Model):
    class Meta:
        table_name = "TABLE_NAME"

    id = UnicodeAttribute(hash_key=True)
    article = MapAttribute()
    # 用lambda确保每次创建新实例时生成当前时间
    created_at = UTCDateTimeAttribute(default_for_new=lambda: datetime.now(timezone.utc))
    last_modified_at = UTCDateTimeAttribute(default_for_new=lambda: datetime.now(timezone.utc))

2. 修改批量操作代码，用更新表达式处理新增/更新：

from pynamodb.expressions.update import UpdateExpression
from pynamodb.expressions.condition import Condition

with ArticleModel.batch_write() as batch:
    # 处理新增/更新
    for identifier, article_data in article_dict["add_or_update"]:
        update_expr = UpdateExpression()
        # 更新文章内容
        update_expr.set(ArticleModel.article, article_data)
        # 每次变更都更新last_modified_at
        update_expr.set(ArticleModel.last_modified_at, datetime.now(timezone.utc))
        # 仅当created_at不存在时（新增条目）才设置它
        update_expr.set(ArticleModel.created_at, datetime.now(timezone.utc), Condition(ArticleModel.created_at.does_not_exist()))
        
        batch.update_item(identifier, update_expression=update_expr)
    
    # 处理删除
    for identifier in article_dict["delete"]:
        batch.delete(ArticleModel(id=identifier))

问题2：条目每次变更时自动更新last_modified_at #

上述代码中，update_expr.set(ArticleModel.last_modified_at, datetime.now(timezone.utc))会在新增或更新条目时，强制将last_modified_at设置为当前时间，直接实现每次变更自动更新的需求。

如果偏好先查询再更新的方式（适合数据量较小的场景），也可以这样写：

with ArticleModel.batch_write() as batch:
    for identifier, article_data in article_dict["add_or_update"]:
        try:
            existing_item = ArticleModel.get(identifier)
            # 保留原created_at，更新内容和修改时间
            item = ArticleModel(
                id=identifier,
                article=article_data,
                created_at=existing_item.created_at,
                last_modified_at=datetime.now(timezone.utc)
            )
            batch.save(item)
        except ArticleModel.DoesNotExist:
            # 新增条目，自动使用默认值设置时间字段
            item = ArticleModel(id=identifier, article=article_data)
            batch.save(item)
    
    for identifier in article_dict["delete"]:
        batch.delete(ArticleModel(id=identifier))

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