如何在YDB的handler函数中为execute_query传递动态参数？

问题 #

需要保留YDB的handler()函数，同时执行含动态数据的SQL查询，但官方示例仅支持静态数据。尝试给execute_query()加参数时触发错误：execute_query() missing 1 required positional argument: 'dynamic_arg'，想知道是否能跳过handler()，或是正确传递参数的方法。

官方示例代码 #

# 创建会话池实例管理YDB会话
pool = ydb.SessionPool(driver)

def execute_query(session):
  # 创建事务并执行查询
  return session.transaction().execute(
    'select 1 as cnt;',
    commit_tx=True,
    settings=ydb.BaseRequestSettings().with_timeout(3).with_operation_timeout(2)
  )

def handler(event, context):
  # 使用retry_operation工具执行查询
  result = pool.retry_operation_sync(execute_query)
  return {
    'statusCode': 200,
    'body': str(result[0].rows[0].cnt == 1),
  }

出错的尝试代码 #

dynamic_arg = somefunc()

def execute_query(session, dynamic_arg):
  # 执行SQL查询
  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)
  )

def handler():
  # 使用retry_operation工具执行查询
  result = pool.retry_operation_sync(execute_query(dynamic_arg))
  return {
    'statusCode': 200,
    'body': str(result[0].rows[0].cnt == 1),
  }

解决方案 #

1. 不能跳过handler函数 #

handler()是YDB Serverless函数的入口，必须保留，无法跳过。

2. 正确传递动态参数的两种方法 #

错误根源是pool.retry_operation_sync需要接收一个能接受session参数的函数，而你直接执行了execute_query(dynamic_arg)，导致提前调用函数并返回结果，而非传递函数本身。可以通过以下两种方式解决：

方法一：使用Lambda表达式包装

def handler(event, context):
    dynamic_arg = somefunc()  # 可根据event/context生成动态参数
    # 用lambda将dynamic_arg绑定到execute_query
    result = pool.retry_operation_sync(lambda session: execute_query(session, dynamic_arg))
    return {
        'statusCode': 200,
        'body': '数据插入完成'
    }

def execute_query(session, dynamic_arg):
    # 禁止用f-string拼接SQL，改用参数绑定避免注入风险
    query = """
        UPSERT INTO tproger (date, engagementRate, reactionsMedian, subscribers, subscriptions, subscriptionsPct, unsubscriptions, unsubscriptionsPct, views, wau)
        VALUES (:date, 1, 2, 3, 4, 5, 6, 7, 8, 9);
    """
    return session.transaction().execute(
        query,
        parameters={'date': dynamic_arg},
        commit_tx=True,
        settings=ydb.BaseRequestSettings().with_timeout(3).with_operation_timeout(2)
    )

方法二：使用functools.partial绑定参数

from functools import partial

def handler(event, context):
    dynamic_arg = somefunc()
    # 用partial预绑定dynamic_arg参数
    bound_func = partial(execute_query, dynamic_arg=dynamic_arg)
    result = pool.retry_operation_sync(bound_func)
    return {
        'statusCode': 200,
        'body': '数据插入完成'
    }

def execute_query(session, dynamic_arg):
    query = """
        UPSERT INTO tproger (date, engagementRate, reactionsMedian, subscribers, subscriptions, subscriptionsPct, unsubscriptions, unsubscriptionsPct, views, wau)
        VALUES (:date, 1, 2, 3, 4, 5, 6, 7, 8, 9);
    """
    return session.transaction().execute(
        query,
        parameters={'date': dynamic_arg},
        commit_tx=True,
        settings=ydb.BaseRequestSettings().with_timeout(3).with_operation_timeout(2)
    )

重要提示：避免SQL注入 #

绝对不要用字符串拼接生成SQL语句，这会引发严重的SQL注入漏洞。YDB支持参数绑定机制，通过parameters字典传递动态值是安全且推荐的做法。

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