如何通过Django将JSON数据存入Snowflake数据表？

我查阅了一些资料，但未找到能解决当前问题的方案。我正在开发一个Django API，期望用户可通过POST请求将数据同时保存到本地数据库和Snowflake数据表。目前我已成功将数据以JSON格式存入本地数据库，示例数据如下：

[
   {
       "id":1,
       "player": "Lilly",
       "power": "Punch"
   }
]

以下是我的views.py文件内容：

class PlayersView(APIView):
    def post(self, request):
        player = PlayerSerializer(data=request.data)
        if player.is_valid():
            player.save()
            snow = Snowflake()
            query = f"""
            create or replace TABLE {table_name} (ID 
            NUMBER(38,0) NOT NULL autoincrement start 0 increment 1, PLAYER 
            VARCHAR(16777216), POWER VARCHAR(16777216)); """
            snow.query(query, return_data=True)
            return Response(player.data, status=status.HTTP_201_CREATED)
        else:
            return Response(player.errors, status=status.HTTP_400_BAD_REQUEST)

    def get(self, request):
        players = Players.objects.all()
        data = PlayerSerializer(players, many=True).data
        print(data)
        return Response(data)

若有疑问，我会进一步说明，提前感谢！

问题分析与解决方案 #

当前代码存在两个核心问题：

1. 每次POST请求都会执行create or replace TABLE，会重复创建/覆盖Snowflake表，导致已有数据丢失
2. 仅实现了建表逻辑，没有将POST的实际数据插入Snowflake表

修改后的代码示例

class PlayersView(APIView):
    def post(self, request):
        player = PlayerSerializer(data=request.data)
        if player.is_valid():
            # 保存数据到本地数据库
            saved_player = player.save()
            
            # 初始化Snowflake连接
            snow = Snowflake()
            
            # 仅在表不存在时创建，避免重复覆盖
            create_table_query = f"""
            create table if not exists {table_name} (
                ID NUMBER(38,0) NOT NULL autoincrement start 0 increment 1, 
                PLAYER VARCHAR(16777216), 
                POWER VARCHAR(16777216)
            );
            """
            snow.query(create_table_query, return_data=True)
            
            # 插入新数据到Snowflake，使用参数化查询避免SQL注入
            insert_query = """
            insert into {table_name} (PLAYER, POWER)
            values (%s, %s);
            """.format(table_name=table_name)
            
            # 从本地保存的对象中获取字段值
            snow.query(insert_query, params=(saved_player.player, saved_player.power), return_data=True)
            
            return Response(player.data, status=status.HTTP_201_CREATED)
        else:
            return Response(player.errors, status=status.HTTP_400_BAD_REQUEST)

    def get(self, request):
        players = Players.objects.all()
        data = PlayerSerializer(players, many=True).data
        print(data)
        return Response(data)

额外注意事项

• 确保table_name变量已正确定义，建议通过配置文件管理，避免硬编码
• 添加异常处理：用try-except包裹Snowflake操作，防止因连接失败、权限不足等问题导致接口报错，同时可根据需求决定是否回滚本地数据库数据
• 性能优化：避免每次请求都新建Snowflake连接，可使用连接池或单例模式复用连接
• 事务一致性：若要求本地数据库与Snowflake操作原子性（要么都成功，要么都失败），需手动实现分布式事务逻辑，比如先执行Snowflake插入，成功后再保存本地数据，失败则回滚

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