使用 SQLAlchemy 在 PostgreSQL 中创建函数和触发器

Question

我使用 SQLAlchemy Engine 创建了一些函数和触发器，但我不想混合使用 Python 和 SQL，所以我为我的 SQL 语句创建了一个单独的文件，我读取了内容并将其传递给 engine.execute()。它不会抛出任何错误，但是函数不是在数据库中创建的，但是如果我通过 pgAdmin 运行相同的 SQL 文件，一切正常。

我的 SQL 文件：

DO $$
BEGIN
  IF NOT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'plpython3u') THEN
    CREATE EXTENSION plpython3u;
  END IF;
END;
$$;

DO $$
BEGIN
  IF NOT EXISTS (SELECT 1 FROM pg_proc WHERE proname = 'my_func') THEN
    CREATE FUNCTION public.my_func() RETURNS TRIGGER LANGUAGE 'plpython3u' NOT LEAKPROOF AS $BODY$
    -- definition
    $BODY$;

    GRANT EXECUTE ON FUNCTION my_func() TO public;
  END IF;
END;
$$;

DO $$
BEGIN
  IF NOT EXISTS (SELECT 1 FROM pg_proc WHERE proname = 'my_func2') THEN
    CREATE FUNCTION public.my_func2() RETURNS TRIGGER LANGUAGE 'plpython3u' NOT LEAKPROOF AS $BODY$
    -- definition
    $BODY$;

    GRANT EXECUTE ON FUNCTION my_func2() TO public;
  END IF;
END;
$$;

我按如下方式运行：

def execute_sql_file(engine, path):
    try:
        with open(path) as file:
            engine.execute(file.read())
    except ProgrammingError:
        raise MyCustomError
    except FileNotFoundError:
        raise MyCustomError

如果我在没有超级用户权限的情况下运行它，它会按预期抛出ProgrammingError。据我了解END; 提交了事务，所以这段代码真的运行了，这些函数应该可供公众使用，但它们甚至没有被创建。欢迎任何想法，谢谢！

Answer 1

我相信您可能混合了BEGIN SQL 命令（Postgresql 扩展）和PL/pgSQL block。 SQL 命令DO 执行一个匿名代码块，就好像它是一个没有参数并返回void 的匿名函数。换句话说，在

DO $$
BEGIN
    ...
END;
$$;

BEGIN / END; 对表示代码块，而不是事务。值得注意的是，从 Postgresql 版本 11 开始 it is possible to manage transactions in a DO block，假设它是 not executed in a transaction block，但其命令是 COMMIT 和 ROLLBACK，而不是关键字 END。

那么问题是你的更改没有提交，尽管你的命令显然被执行了——正如错误所证明的那样，如果没有以合适的权限运行。这个问题是由how SQLAlchemy autocommit feature works 引起的。简而言之，它会检查您的语句/命令并尝试确定它是数据更改操作还是 DDL 语句。这适用于INSERT、DELETE、UPDATE 等基本操作，但并不完美。事实上，它不可能总是正确地确定语句是否更改了数据；例如SELECT my_mutating_procedure()就是这样一个声明。因此，如果进行更复杂的操作，它需要一些帮助。一种方法是通过将 SQL 字符串包装在 text() 构造和 using execution_options() 中来指示自动提交机制应该提交：

engine.execute(text("SELECT my_mutating_procedure()").
               execution_options(autocommit=True))

还可以使用DDL 构造显式指示 SQLAlchemy 该命令是文字 DDL 语句：

from sqlalchemy.schema import DDL

def execute_sql_file(engine, path):
    try:
        with open(path) as file:
            stmt = file.read()

        # Not strictly DDL, but a series of DO commands that execute DDL
        ddl_stmt = DDL(stmt)
        engine.execute(ddl_stmt)

    except ProgrammingError:
        raise MyCustomError

    except FileNotFoundError:
        raise MyCustomError

至于为什么它与 pgAdmin 一起工作，如果没有引发错误，它可能默认提交。