如何用FastAPI发送POST请求实现Angular Ag-Grid编辑并同步Snowflake数据

一、前端Angular部分 #

1. 配置Ag-Grid可编辑列 #

先确保列配置中开启编辑权限，主键列建议设为不可编辑：

columnDefs = [
  { field: 'id', editable: false, headerName: 'ID' },
  { field: 'name', editable: true, headerName: '名称' },
  { field: 'value', editable: true, headerName: '数值' }
];
rowData: any[] = []; // 从数据库加载的初始数据

2. 捕获编辑变更 #

在模板中绑定cellValueChanged事件，收集修改后的行数据：

<ag-grid-angular
  style="width: 100%; height: 500px;"
  class="ag-theme-alpine"
  [rowData]="rowData"
  [columnDefs]="columnDefs"
  (cellValueChanged)="onCellValueChanged($event)"
></ag-grid-angular>
<button (click)="saveChanges()">保存变更</button>

组件类中维护变更列表，避免重复添加：

import { Component } from '@angular/core';
import { CellValueChangedEvent } from 'ag-grid-community';
import { DataSyncService } from '../services/data-sync.service';

@Component({
  selector: 'app-grid',
  templateUrl: './grid.component.html',
  styleUrls: ['./grid.component.css']
})
export class GridComponent {
  changedRows: any[] = [];

  constructor(private dataSyncService: DataSyncService) {}

  onCellValueChanged(event: CellValueChangedEvent) {
    const updatedRow = event.data;
    const existingIndex = this.changedRows.findIndex(row => row.id === updatedRow.id);
    if (existingIndex > -1) {
      this.changedRows[existingIndex] = updatedRow;
    } else {
      this.changedRows.push(updatedRow);
    }
  }

  saveChanges() {
    if (this.changedRows.length === 0) {
      alert('没有变更需要保存');
      return;
    }
    this.dataSyncService.syncChanges(this.changedRows)
      .subscribe({
        next: () => {
          alert('保存成功');
          this.changedRows = [];
        },
        error: (err) => {
          console.error('保存失败:', err);
          alert('保存失败，请查看控制台日志');
        }
      });
  }
}

3. 创建HTTP同步服务 #

先在app.module.ts中导入HttpClientModule，再创建服务发送POST请求：

// src/app/services/data-sync.service.ts
import { Injectable } from '@angular/core';
import { HttpClient } from '@angular/common/http';

@Injectable({
  providedIn: 'root'
})
export class DataSyncService {
  private apiUrl = 'http://localhost:5000/sync-data'; // Python后端地址

  constructor(private http: HttpClient) {}

  syncChanges(changedRows: any[]) {
    return this.http.post(this.apiUrl, { rows: changedRows });
  }
}

二、后端Python部分 #

1. 安装依赖 #

执行命令安装所需包：

pip install flask flask-cors snowflake-connector-python

2. 编写Flask API处理请求 #

创建app.py文件，实现POST接口并连接Snowflake执行更新：

from flask import Flask, request, jsonify
from flask_cors import CORS
import snowflake.connector

app = Flask(__name__)
CORS(app)  # 允许跨域请求

# 替换为你的Snowflake实际配置
SNOWFLAKE_CONFIG = {
    'user': '你的用户名',
    'password': '你的密码',
    'account': '你的账户标识（如xxx.us-west-2）',
    'warehouse': '你的仓库名',
    'database': '你的数据库名',
    'schema': '你的模式名'
}

@app.route('/sync-data', methods=['POST'])
def sync_data():
    try:
        request_data = request.get_json()
        changed_rows = request_data.get('rows', [])
        
        if not changed_rows:
            return jsonify({'status': 'success', 'msg': '无变更数据'}), 200
        
        # 连接Snowflake并执行批量更新
        conn = snowflake.connector.connect(**SNOWFLAKE_CONFIG)
        cursor = conn.cursor()
        
        for row in changed_rows:
            update_sql = """
                UPDATE 你的表名
                SET NAME = %s, VALUE = %s
                WHERE ID = %s
            """
            cursor.execute(update_sql, (row['name'], row['value'], row['id']))
        
        conn.commit()
        cursor.close()
        conn.close()
        
        return jsonify({'status': 'success', 'msg': f'成功更新{len(changed_rows)}条数据'}), 200
    
    except Exception as e:
        return jsonify({'status': 'error', 'msg': str(e)}), 500

if __name__ == '__main__':
    app.run(debug=True)

3. 启动后端服务 #

执行命令启动Flask应用：

python app.py

关键注意事项 #

• 替换代码中所有你的xxx为实际的Snowflake配置、表名等信息
• Snowflake用户需拥有目标表的UPDATE权限
• 生产环境禁止硬编码敏感信息（如密码），建议用环境变量存储
• 前端启动后需确保与后端端口通信正常（示例中后端用5000端口）

内容的提问来源于stack exchange，提问作者Tenzing Sherpa