使用 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

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

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

的BEGIN/END;对表示所述码块，而不是一个交易。值得注意的是，从 Postgresql 版本 11 开始，可以管理DO块中的事务，因为它不在事务块中执行，但执行该操作的命令是COMMITand ROLLBACK，而不是关键字END。

问题是您的更改没有提交，尽管您的命令显然已执行 - 正如错误所证明的那样，如果没有以合适的权限运行。此问题是由SQLAlchemy 自动提交功能的工作方式引起的。简而言之，它会检查您的语句/命令并尝试确定它是数据更改操作还是 DDL 语句。这适用于INSERT、DELETE、等基本操作UPDATE，但并不完美。事实上，它不可能总是正确地判断一个语句是否改变了数据；例如SELECT my_mutating_procedure()是这样的声明。所以它需要一些帮助，如果做更复杂的操作。一种方法是通过将 SQL 字符串包装在using 中来指示自动提交机制它应该提交：text()结构中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 一起工作，如果没有出现错误，它可能默认提交。