使用 Alembic 的存储过程：MySQL 语法错误

Question

我正在使用 Alembic 将存储过程加载到 MySQL 数据库中。在 cookbook 之后，我能够创建创建和删除过程所需的对象。现在我想升级版本以实际加载过程，我在过程定义所需的 DELIMITER $$ 附近得到 SQL 语法。我还尝试删除 DELIMITER 并替换 AS 以启动程序，但这也没有用。我什至尝试了食谱中的简单示例函数，但这也在 $ 附近给了我同样的语法错误。

class ReplaceableObject:
    def __init__(self, name, sqltext):
        self.name = name
        self.sqltext = sqltext


class ReversibleOp(MigrateOperation):
    def __init__(self, target):
        self.target = target

    @classmethod
    def invoke_for_target(cls, operations, target):
        op = cls(target)
        return operations.invoke(op)

    def reverse(self):
        raise NotImplementedError()

    @classmethod
    def _get_object_from_version(cls, operations, ident):
        version, objname = ident.split(".")

        module = operations.get_context().script.get_revision(version).module
        obj = getattr(module, objname)
        return obj

    @classmethod
    def replace(cls, operations, target, replaces=None, replace_with=None):

        if replaces:
            old_obj = cls._get_object_from_version(operations, replaces)
            drop_old = cls(old_obj).reverse()
            create_new = cls(target)
        elif replace_with:
            old_obj = cls._get_object_from_version(operations, replace_with)
            drop_old = cls(target).reverse()
            create_new = cls(old_obj)
        else:
            raise TypeError("replaces or replace_with is required")

        operations.invoke(drop_old)
        operations.invoke(create_new)


@Operations.register_operation("create_sp", "invoke_for_target")
@Operations.register_operation("replace_sp", "replace")
class CreateSPOp(ReversibleOp):
    def reverse(self):
        return DropSPOp(self.target)


@Operations.register_operation("drop_sp", "invoke_for_target")
class DropSPOp(ReversibleOp):
    def reverse(self):
        return CreateSPOp(self.target)


@Operations.implementation_for(CreateSPOp)
def create_sp(operations, operation):
    operations.execute(
        """
        DELIMITER $$
        CREATE PROCEDURE %s %s
        DELIMITER ;
        """ % (
            operation.target.name, operation.target.sqltext
        )
    )


@Operations.implementation_for(DropSPOp)
def drop_sp(operations, operation):
    operations.execute("DROP PROCEDURE IF EXISTS %s" % operation.target.name)

这就是我尝试创建它的方式：

my_procedure = ReplaceableObject(
    "my_procedure(IN arg1 TEXT, IN arg2 TEXT, OUT res TEXT)",
    """
    BEGIN
        -- procedure implementation
    END $$   
    """
)

def upgrade():
    op.create_sp(my_procedure)

def downgrade():
    op.drop_sp(my_procedure)

这是错误（我使用 pymysql 作为客户端）：

sqlalchemy.exc.ProgrammingError: (pymysql.err.ProgrammingError) (1064, "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'DELIMITER $$\n        CREATE PROCEDURE my_procedure(IN arg1 TEXT, IN arg2 TEXT, OU' at line 1")

Answer 1

您不需要 DELIMITER，除非在支持一系列以分号分隔的语句的环境中 — MySQL 客户端。

其实DELIMITER命令是mysql client builtin，MySQL服务器无法识别。

通过 API 向 MySQL 服务器提交语句时不得使用 DELIMITER，即使它是 CREATE PROCEDURE.

使用 Alembic 的存储过程：MySQL 语法错误

Stored procedure with Alembic: MySQL Syntax Error

python

mysql

sqlalchemy

sqlalchemy-migrate

alembic