Dank des Vorschlags von Michael Bayer , habe ich eine Lösung für meine eigene Frage gefunden:Der Trick besteht darin, zu verwenden Compiler von SQLAlchemy , und etwas richtiges Escaping:
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://[email protected]/mytest')
print(call(engine, 'test_f', a='a').scalar())
print(call(engine, 'test_f', b='b').scalar())
