python - 使用 SQLAlchemy 的数据库函数中的命名参数

Question

我的数据库 (Postgres) 中有一个如下所示的函数:

create function test_f(a text default '*', b text default '+') returns text as $$
    select a || ' ' || b;
$$ language sql;

Postgres 允许使用命名参数调用它:

mytest=> select test_f('a', 'b');                                           
test_f                                                                      
--------                                                                    
a b                                                                         
(1 row)                                                                     

mytest=> select test_f('a');                                                
test_f                                                                      
--------                                                                    
a +                                                                         
(1 row)                                                                     

mytest=> select test_f(b:='a');                                             
test_f                                                                      
--------                                                                    
* a                                                                         
(1 row)                                                                     

我想用 Python 做同样的事情，使用 SQLAlchemy's func construct ，但似乎 func 不支持命名参数:

In [85]: print(sqlalchemy.func.test_f('a', 'b'))                            
test_f(:test_f_1, :test_f_2)                                                

In [86]: print(sqlalchemy.func.test_f('a'))                                 
test_f(:test_f_1)                                                           

In [87]: print(sqlalchemy.func.test_f(a='a'))                               
test_f()                                                                    

我是不是遗漏了什么，或者 func 不支持命名参数？

Answer 1

感谢 suggestion of Michael Bayer ，我想出了一个解决我自己问题的方法:诀窍是使用 SQLAlchemy's compiler ，以及一些适当的转义:

from psycopg2.extensions import adapt as sqlescape                          
import sqlalchemy                                                           
from sqlalchemy import select                                               
from sqlalchemy.ext.compiler import compiles                                
from sqlalchemy.sql.expression import ColumnClause                          


class MyFunc(ColumnClause):                                                 
    def __init__(self, *args, **kwargs):                                    
        self.kwargs = kwargs                                                
        super().__init__(*args)                                             


@compiles(MyFunc)                                                           
def compile_myfunc(element, compiler, **kw):                                
    s = ','.join("%s:=%s" % (k, sqlescape(v)) for k, v in element.kwargs.items())
    return "%s(%s)" % (element.name, s)                                     


def call(engine, func, **kwargs):                                           
    return engine.execute(select([MyFunc(func, **kwargs)]))                 


engine = sqlalchemy.create_engine('postgresql+psycopg2://lbolla@localhost/mytest')
print(call(engine, 'test_f', a='a').scalar())                               
print(call(engine, 'test_f', b='b').scalar())