如何在自定义编译表达式中使用bindparam()?

2023-12-05

我的代码基于 @zzzeeek 的回答这个问题。 我对其进行了一些扩展,因此它考虑了 Postgresql 的 NULL 和 ARRAY。

class values(FromClause):
    named_with_column = True

    def __init__(self, columns, *args, **kw):
        self._column_args = columns
        self.list = args
        self.alias_name = self.name = kw.pop('alias_name', None)

    def _populate_column_collection(self):
        # self._columns.update((col.name, col) for col in self._column_args)
        for c in self._column_args:
        c._make_proxy(self, c.name)


@compiles(values)
def compile_values(element, compiler, asfrom=False, **kw):
    columns = element.columns
    v = "VALUES %s" % ", ".join(
    "(%s)" % ", ".join(
        ((compiler.visit_array(elem)+'::'+str(column.type)) if isinstance(column.type, ARRAY) else
         compiler.render_literal_value(elem, column.type))
        if elem is not None else compiler.render_literal_value(elem, NULLTYPE)
        for elem, column in zip(tup, columns))
    for tup in element.list
    )
    if asfrom:
        if element.alias_name:
            v = "(%s) AS %s (%s)" % (v, element.alias_name, (", ".join(c.name for c in element.columns)))
        else:
            v = "(%s)" % v
    return v

一切都工作正常,直到事实证明我无法将带有“%”符号的值插入到这个 VALUES 子句中 - 它们被内联到结果语句中,这似乎会导致绑定问题

我想如果不是render_literal_value()我们用了bindparam()我们可以避免这样的错误。但一切都在@compiles应该返回纯文本,对吗?我如何修改它以获得基于参数的查询?


我得到了它。 Bindparams 的实际值保存在一个名为SQLCompiler这通常是特定于方言的。这里(链接到 GitHub)是bindparams存储在的地方SQLCompiler查询编译过程中的实例

所以我的代码片段的最终版本如下所示:

class values(FromClause):
    named_with_column = True

    def __init__(self, columns, *args, **kw):
        self._column_args = columns
        self.list = args
        self.alias_name = self.name = kw.pop('alias_name', None)

    def _populate_column_collection(self):
        # self._columns.update((col.name, col) for col in self._column_args)
        for c in self._column_args:
            c._make_proxy(self, c.name)


@compiles(values)
def compile_values(clause, compiler, asfrom=False, **kw):
    def decide(value, column):
        add_type_hint = False
        if isinstance(value, array) and not value.clauses:  # for empty array literals
            add_type_hint = True

        if isinstance(value, ClauseElement):
            intermediate = compiler.process(value)
            if add_type_hint:
                intermediate += '::' + str(column.type)
            return intermediate

        elif value is None:
            return compiler.render_literal_value(
                value,
                NULLTYPE
            ) + '::' + str(column.type)
        else:
            return compiler.process(
                bindparam(
                    None,
                    value=compiler.render_literal_value(
                        value,
                        column.type
                    ).strip("'")
                )
            ) + '::' + str(column.type)

    columns = clause.columns
    v = "VALUES %s" % ", ".join(
        "(%s)" % ", ".join(
            decide(elem, column)
            for elem, column in zip(tup, columns))
        for tup in clause.list
    )
    if asfrom:
        if clause.alias_name:
            v = "(%s) AS %s (%s)" % (v, clause.alias_name, (", ".join(c.name for c in clause.columns)))
        else:
            v = "(%s)" % v
    return v
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

如何在自定义编译表达式中使用bindparam()? 的相关文章

随机推荐