Flask应用Azure Blob存储上传及SQL Server插入失败求助

问题概述 #

开发Flask应用时，upload_file函数存在两个核心故障：

1. 文件无法正确上传至Azure Blob Storage
2. SQL Server数据库数据插入失败

核心代码 #

数据库与Azure配置 #

# 数据库凭据
SERVER = ""
DATABASE = ""
USERNAME = ""
PASSWORD = ""
DRIVER = ""

# 创建连接字符串
connection_string = f'DRIVER={{{DRIVER}}};SERVER={SERVER};DATABASE={DATABASE};UID={USERNAME};PWD={PASSWORD}'

# Azure Blob Storage配置
connect_str = ""
container_name = 'files'
blob_service_client = BlobServiceClient.from_connection_string(connect_str)

上传核心函数 #

@app.route('/upload', methods=['POST'])
def upload_file():
    file = request.files['file']
    if not file:
        return jsonify({'message': 'No file provided'}), 400

    blob_path = f"upload/trainfile/{file.filename}"
    blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_path)
    
    conn = None
    try:
        if blob_client.exists():
            return jsonify({'message': 'File already exists'}), 409

        blob_client.upload_blob(file)
        pdf_url = blob_client.url

        conn = pyodbc.connect(connection_string)
        cursor = conn.cursor()

        main_fields = {key: request.form[key] for key in request.form.keys() if not key.startswith('item_')}
        columns = ', '.join(main_fields.keys())
        placeholders = ', '.join('?' * len(main_fields))
        main_sql = f"INSERT INTO TrainData ({columns}, pdf_url, created_at) VALUES ({placeholders}, ?, GETDATE())"

        cursor.execute(main_sql, *(list(main_fields.values()) + [pdf_url]))
        train_data_id = cursor.execute("SELECT @@IDENTITY AS id;").fetchval()

        item_keys = [key for key in request.form.keys() if key.startswith('item_')]
        grouped_item_data = {}
        for key in item_keys:
            index, item_attr = key.split('_')[1], '_'.join(key.split('_')[2:])
            grouped_item_data.setdefault(index, {})[item_attr] = request.form[key]

        for item in grouped_item_data.values():
            item_columns = ', '.join(item.keys())
            item_placeholders = ', '.join('?' * len(item))
            item_sql = f"INSERT INTO ItemDetails (train_data_id, {item_columns}) VALUES (?, {item_placeholders})"
            cursor.execute(item_sql, train_data_id, *item.values())

        conn.commit()
        cursor.close()
        return jsonify({'message': 'Data saved successfully', 'file_url': pdf_url}), 200
    except Exception as e:
        if conn:
            conn.rollback()

        print("An error occurred:", e)
        return jsonify({'message': 'Failed to save data', 'error': str(e)}), 500
    finally:
        if conn:
            conn.close()

排查与修复方案 #

Azure Blob Storage 部分 #

• 验证连接字符串：确保connect_str已填入Azure存储账户的完整连接字符串（从Azure门户→存储账户→访问密钥中复制），不能为空。
• 修复文件流上传：Flask的FileStorage对象可能因表单读取导致指针偏移，上传前重置指针：

  file.seek(0)
  blob_client.upload_blob(file)
  

• 细化Blob上传异常捕获：单独捕获Blob上传错误，定位是否为权限/网络问题：

  try:
      file.seek(0)
      blob_client.upload_blob(file)
      pdf_url = blob_client.url
  except Exception as blob_err:
      print("Blob上传失败:", blob_err)
      return jsonify({'message': '文件上传失败', 'error': str(blob_err)}), 500
  

• 容器权限检查：确认files容器的访问权限至少为Blob级别，或使用账户密钥/SAS令牌授权写入。

SQL Server 部分 #

• 替换@@IDENTITY为SCOPE_IDENTITY()：@@IDENTITY可能返回触发器生成的其他ID，改用SCOPE_IDENTITY()获取当前会话插入的主键：

  train_data_id = cursor.execute("SELECT SCOPE_IDENTITY() AS id;").fetchval()
  

• 检查参数与列名匹配：确保main_fields的键与TrainData表列名完全一致，item_字段拆分逻辑正确（例如item_0_name需拆分为index=0、item_attr=name）。
• 添加Blob回滚逻辑：若数据库插入失败，删除已上传的Blob，避免数据不一致：

  except Exception as e:
      if conn:
          conn.rollback()
      # 删除已上传的Blob
      try:
          blob_client.delete_blob()
      except:
          pass
      print("数据库操作失败:", e)
      return jsonify({'message': '数据保存失败', 'error': str(e)}), 500
  

• 验证数据库连接字符串：确保DRIVER使用最新版本（如{ODBC Driver 18 for SQL Server}），且所有凭据填写完整。

SQLAlchemy 适配建议 #

若使用SQLAlchemy，简化连接与事务管理，避免手动操作游标：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import urllib.parse

engine = create_engine("mssql+pyodbc:///?odbc_connect=" + urllib.parse.quote_plus(connection_string))
Session = sessionmaker(bind=engine)
session = Session()

# 插入主数据
train_data = TrainData(**main_fields, pdf_url=pdf_url)
session.add(train_data)
session.commit()
train_data_id = train_data.id

# 插入子数据
for item in grouped_item_data.values():
    item_detail = ItemDetails(train_data_id=train_data_id, **item)
    session.add(item_detail)
session.commit()
session.close()

通用调试建议 #

• 打印main_fields、grouped_item_data的内容，确认表单数据是否正确接收。
• 单独测试Blob上传和数据库插入功能，排除单一模块故障后再整合。

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