本文详解如何在 ydb 的 `retry_operation_sync` 机制中安全、灵活地向内层查询函数传递动态参数，避免 sql 注入风险，并提供闭包封装与参数化查询两种专业方案。
在使用 YDB Python SDK 时，pool.retry_operation_sync() 要求传入的函数签名必须严格为 func(session) —— 即仅接收一个 session 参数。因此，直接在 execute_query(session, dynamic_arg) 中添加额外参数会导致 TypeError: missing 1 required positional argument 错误。这不是限制，而是设计使然：它强制开发者将外部上下文（如动态值）通过闭包或参数化查询方式安全注入，而非破坏函数契约。
✅ 方案一：使用闭包封装动态参数（推荐用于简单场景）
通过外层工厂函数 prepare_execute_query(dynamic_arg) 返回一个符合 session 单参签名的内层函数，实现参数“预绑定”：
dynamic_arg = somefunc() # 如 datetime.now().isoformat()
def prepare_execute_query(dynamic_arg):
def execute_query(session):
return session.transaction().execute(
f"""
UPSERT INTO tproger (
date, engagementRate, reactionsMedian, subscribers,
subscriptions, subscriptionsPct, unsubscriptions,
unsubscriptionsPct, views, wau
) VALUES ({dynamic_arg}, 1, 2, 3, 4, 5, 6, 7, 8, 9);
""",
commit_tx=True,
settings=ydb.BaseRequestSettings()
.with_timeout(3)
.with_operation_timeout(2)
)
return execute_query
def handler(event, context):
result = pool.retry_operation_sync(prepare_execute_query(dynamic_arg))
return {
'statusCode': 200,
'body': 'OK'
}⚠️ 注意：此方式虽简洁，但存在严重 SQL 注入风险——若 dynamic_arg 来自用户输入（如 HTTP 请求体、路径参数），拼接字符串将导致漏洞。生产环境严禁直接插值未校验的变量。
✅ 方案二：使用参数化查询（强烈推荐，生产必备）
YDB 原生支持带命名/位置参数的预编译查询（Prepared Query），既安全又高效。应始终优先采用此方式：
def execute_query_with_params(session, dynamic_arg):
return session.transaction().execute(
"""
UPSERT INTO tproger (
date, engagementRate, reactionsMedian, subscribers,
subscriptions, subscriptionsPct, unsubscriptions,
unsubscriptionsPct, views, wau
) VALUES ($date, 1, 2, 3, 4, 5, 6, 7, 8, 9);
""",
commit_tx=True,
parameters={
'$date': ydb.PrimitiveType.String(dynamic_arg) # 类型需显式声明
},
settings=ydb.BaseRequestSettings()
.with_timeout(3)
.with_operation_timeout(2)
)
def handler(event, context):
dynamic_arg = somefunc() # 安全：参数由 YDB 驱动处理，不参与 SQL 解析
result = pool.retry_operation_sync(
lambda session: execute_query_with_params(session, dynamic_arg)
)
return {'statusCode': 200, 'body': 'UPSERT completed'}✅ 优势总结：
- 零 SQL 注入风险：参数经驱动序列化后作为独立 payload 传输，与 SQL 结构完全隔离；
- 类型安全：ydb.PrimitiveType.* 显式声明类型，避免隐式转换错误；
- 性能更优：YDB 服务端可缓存预编译计划，重复执行更快；
- 兼容性好：支持所有 YDB 数据类型（Uint64, Timestamp, Decimal, JSON 等）。
? 总结
永远不要用字符串格式化（f"" / % / .format()）拼接用户可控数据到 SQL 中。正确路径是：
- 使用 lambda session: func(session, arg) 或闭包工厂函数满足 retry_operation_sync 签名要求；
- 核心原则：所有动态值必须通过 parameters={} 以参数化方式传入 execute()；
- 务必为每个参数指定准确的 YDB 类型（如 ydb.PrimitiveType.Int64(val)），尤其注意时间、浮点、大数等易出错类型。
遵循以上实践，即可在保持 YDB SDK 最佳重试语义的同时，安全、可靠、高性能地执行带动态参数的 SQL 操作。
