YDB 中动态参数安全传入 SQL 查询的正确实践

本文详解如何在 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 中。正确路径是：

1. 使用 lambda session: func(session, arg) 或闭包工厂函数满足 retry_operation_sync 签名要求；
2. 核心原则：所有动态值必须通过 parameters={} 以参数化方式传入 execute()；
3. 务必为每个参数指定准确的 YDB 类型（如 ydb.PrimitiveType.Int64(val)），尤其注意时间、浮点、大数等易出错类型。

遵循以上实践，即可在保持 YDB SDK 最佳重试语义的同时，安全、可靠、高性能地执行带动态参数的 SQL 操作。