YDB 中如何安全地将动态参数传递给会话池执行的 SQL 查询_技术教程

本文讲解在 ydb python sdk 中，如何通过闭包或参数化查询方式，将动态值安全、正确地传入 `session.transaction().execute()`，避免 sql 注入与参数绑定错误。

在使用 YDB 的 SessionPool 执行数据库操作时，retry_operation_sync() 要求传入一个单参数函数（仅接收 session），因此无法直接向 execute_query(session, dynamic_arg) 这类多参数函数传递动态数据。常见错误是误将函数调用结果（如 execute_query(dynamic_arg)）传给 retry_operation_sync()，导致类型不匹配或提前执行。

✅ 正确做法是：利用 Python 闭包封装动态参数，返回符合签名要求的单参函数：

dynamic_arg = somefunc()  # 例如：datetime.now().isoformat()

def prepare_execute_query(arg_value):
    def execute_query(session):
        return session.transaction().execute(
            f"""
            UPSERT INTO tproger (
                date, engagementRate, reactionsMedian,
                subscribers, subscriptions, subscriptionsPct,
                unsubscriptions, unsubscriptionsPct, views, wau
            ) VALUES (
                '{arg_value}', 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'
    }

⚠️ 但注意：上述字符串拼接方式存在严重风险——若 dynamic_arg 来自用户输入或外部系统，极易引发 SQL 注入攻击（例如 arg_value = "2025-01-01'; DROP TABLE tproger; --"）。

✅ 推荐方案：使用参数化查询（Prepared Query），由 YDB 服务端安全绑定参数：

def execute_query_with_params(session):
    # 预编译带命名参数的语句（推荐在初始化阶段复用）
    query = """
    UPSERT INTO tproger (
        date, engagementRate, reactionsMedian,
        subscribers, subscriptions, subscriptionsPct,
        unsubscriptions, unsubscriptionsPct, views, wau
    ) VALUES (
        $date, $engagement, $reactions, $subs, $subspct,
        $unsubs, $unsubspct, $views, $wau
    );
    """

    dynamic_arg = somefunc()  # 动态获取值

    # 构造参数字典（键名需与 $xxx 一致，类型需匹配表定义）
    params = {
        '$date': dynamic_arg,
        '$engagement': 1.0,
        '$reactions': 2.0,
        '$subs': 3,
        '$subspct': 4.0,
        '$unsubs': 6,
        '$unsubspct': 7.0,
        '$views': 8,
        '$wau': 9
    }

    return session.transaction().execute(
        query,
        parameters=params,
        commit_tx=True,
        settings=ydb.BaseRequestSettings()
            .with_timeout(3)
            .with_operation_timeout(2)
    )

def handler(event, context):
    result = pool.retry_operation_sync(execute_query_with_params)
    return {'statusCode': 200, 'body': 'Upsert completed'}

? 关键要点总结：