需求:使用Python开发Azure Function实现Azure SQL数据翻译更新
Python Azure Function:Azure SQL 数据翻译并更新方案
前置准备
- 安装必要依赖包:
pip install azure-functions pyodbc google-cloud-translate - 配置Azure资源:
- 确保Azure SQL数据库允许Azure Functions的IP访问(或启用服务端点)
- 在Azure Function的应用设置中添加:
AZURE_SQL_CONNECTION_STRING:Azure SQL的连接字符串GOOGLE_APPLICATION_CREDENTIALS:Google服务账号密钥文件的路径(部署时可通过应用设置注入,本地开发可直接指定路径)
- 配置Google Cloud:
- 启用Cloud Translation API,创建服务账号并下载密钥文件
函数实现(以TimerTrigger为例)
以下是定时触发的完整代码,实现从SQL提取数据、翻译、更新的流程:
import azure.functions as func import pyodbc import os from google.cloud import translate_v2 as translate def main(mytimer: func.TimerRequest) -> None: # 初始化翻译客户端 translate_client = translate.Client() # 连接Azure SQL数据库 conn_str = os.getenv("AZURE_SQL_CONNECTION_STRING") conn = pyodbc.connect(conn_str) cursor = conn.cursor() try: # 提取待翻译数据(建议添加过滤条件避免重复翻译,比如新增is_translated标记) cursor.execute("SELECT id, verbatim FROM your_table WHERE verbatim IS NOT NULL AND verbatim != ''") rows = cursor.fetchall() for row in rows: record_id, original_text = row # 调用Google Translate API翻译为英文 result = translate_client.translate(original_text, target_language='en') translated_text = result['translatedText'] # 更新回原表 update_query = """ UPDATE your_table SET verbatim = ? WHERE id = ? """ cursor.execute(update_query, translated_text, record_id) conn.commit() print(f"成功处理 {len(rows)} 条记录") except Exception as e: print(f"处理过程出错: {str(e)}") conn.rollback() finally: cursor.close() conn.close()
关键注意事项
- 数据过滤:建议新增
is_translated布尔列,查询时仅提取未翻译数据,避免重复操作浪费资源 - 批量处理:数据量较大时,改用批量查询+批量更新逻辑,减少数据库交互次数,提升性能
- 权限配置:Azure Function的服务主体需拥有Azure SQL的
db_datareader和db_datawriter权限;Google服务账号需具备Cloud Translation API的调用权限 - 错误处理:完善异常捕获逻辑,确保数据库连接在出错时正常关闭,避免连接泄漏
部署说明
- 按照Azure Function项目结构整理代码(包含
function.json配置触发规则) - 通过Azure CLI或门户将代码部署到Function App
- 在Function App的应用设置中配置好
AZURE_SQL_CONNECTION_STRING和GOOGLE_APPLICATION_CREDENTIALS(密钥文件可通过Azure机密管理注入)
内容的提问来源于stack exchange,提问作者Vipinkumar Jha
相关产品推荐
相关产品推荐

