FastAPI集成Boto3操作DynamoDB时获取旧数据的问题及解决方法

核心原因 #

1. DynamoDB二级索引的最终一致性限制：你使用的application_id-index属于全局二级索引（GSI），DynamoDB的GSI仅支持最终一致性读取——写入操作完成后，索引数据同步存在延迟，此时立刻查询索引会返回未更新的旧数据。重启服务器后，索引已完成同步，因此能获取到最新内容。
2. 潜在的进程内缓存：如果get_dict_items函数或其上游调用逻辑被添加了缓存装饰器（比如functools.lru_cache），会导致查询结果被缓存，重启服务器后缓存失效才会返回新数据。另外，若全局的dynamodb资源对象（boto3.resource('dynamodb')）存在元数据缓存，也可能间接影响，但概率较低。

解决办法 #

针对索引一致性问题 #

• 改用主键强一致性查询主表：如果业务逻辑允许，直接通过主键查询主表，主表支持ConsistentRead=True参数开启强一致性读取，确保拿到最新数据。示例代码修改：

  def get_dict_items(table_name, app_list):  # 避免用dict作为参数名，与内置类型冲突
      try:
          # 每次查询重新创建DynamoDB资源
          dynamodb = boto3.resource('dynamodb')
          table = dynamodb.Table(table_name)
          result = []
          # 假设主键字段为id，根据实际主键调整
          for app in app_list:
              response = table.get_item(
                  Key={'id': app['id']},
                  ConsistentRead=True
              )
              result.append(response.get("Item", []))
          return 200, result
      except ClientError as ce:
          return 500, f"Error getting items from table {table_name}: {ce}"
      except (BotoCoreError, NoCredentialsError) as e:
          return 500, f"Unexpected error getting items from table {table_name}: {e}"
  

• 若必须使用二级索引：
  • 非实时场景下，可在写入操作后添加短暂等待（如0.5-1秒）再执行查询，但不适合高并发或强实时需求。
  • 实时场景可借助DynamoDB Streams监听写入事件，在事件触发后再执行查询，确保索引已完成同步。

排查缓存问题 #

• 检查缓存装饰器：确认get_dict_items或其调用的上游函数没有使用lru_cache等缓存装饰器，若有则移除或调整缓存失效策略。
• 彻底隔离全局资源：将boto3.resource('dynamodb')的初始化逻辑移到函数内部，避免全局资源的缓存影响（你之前仅移动了Table初始化，现在需把整个资源创建也移进去）。

验证步骤 #

1. 先尝试用主键强一致性查询主表，如果能拿到最新数据，即可确认是二级索引的最终一致性导致的问题。
2. 临时移除函数可能存在的缓存装饰器，测试查询是否恢复正常。

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