在事件侦听器中创建数据库触发器

Question

我有一个烧瓶API，我使用烧瓶SQLAlChemy来处理SQLite数据库。我有一个存储日志条目的表，我想将最大行数限制为n。因为插入也是使用原始SQL从烧瓶之外的另一个脚本进行的，所以我创建了一个触发器来检查行数，如果行数高于n，则删除最旧的行数：

CREATE TRIGGER 'trigger_log_insert' 
BEFORE INSERT ON 'connection_logs' 
WHEN ( SELECT count(*) FROM  'connection_logs' ) > 5 
BEGIN 
DELETE FROM 'connection_logs' 
WHERE id NOT IN ( SELECT id FROM 'connection_logs' ORDER BY id DESC LIMIT 5 ); 
END

此触发器按预期工作，但我正在努力使用烧瓶-sqlalChemy设置它。如何设置触发器/执行原始SQL使用烧瓶sqlalChemy？SQL只需要在db创建后执行一次，所以我打算在create_all（）语句后立即执行它。我偶然发现了这个StackOverflow的答案，它提出了一个显然很快就会被弃用的解决方案。我还阅读了关于自定义DDL的SQLAlChemy留档，但是我不知道如何使用flask_sqlalchemy创建这个自定义DDL。当我在SQLAlChemy留档中创建DDL时，我会收到一条错误消息，说DDL对象没有绑定到引擎或连接。

trigger = DDL(
"CREATE TRIGGER 'trigger_log_insert'"
"BEFORE INSERT ON 'connection_logs'"
"WHEN ( SELECT count(*) FROM  'connection_logs' ) > 5"
"BEGIN"
"DELETE FROM 'connection_logs' WHERE id NOT IN"
"("
"SELECT id FROM 'connection_logs' ORDER BY id DESC LIMIT 5"
");"
"END"
)

event.listen(ConnectionLog, 'after_create', trigger.execute())

我的模型是使用flask sqlalchemy的声明性基本模型定义的：

class ConnectionLog(db.Model):
    __tablename__ = 'connection_logs'

Answer 1

您不需要创建DDL实例，您可以在监听器函数中执行SQL。相关的文档在这里。

import sqlalchemy as sa
...

class ConnectionLog(db.Model):
    __tablename__ = 'connection_logs'
...

def after_create(target, connection, **kw):
    connection.execute(sa.text("""\
        CREATE TRIGGER 'trigger_log_insert'
        BEFORE INSERT ON 'connection_logs'
        WHEN ( SELECT count(*) FROM  'connection_logs' ) > 5
        BEGIN
        DELETE FROM 'connection_logs' WHERE id NOT IN
        (
        SELECT id FROM 'connection_logs' ORDER BY id DESC LIMIT 5
        );
        END
        """
    ))

# Listen on the underlying table object, not on the model class.
sa.event.listen(ConnectionLog.__table__, "after_create", after_create)

确保解释器在创建表之前已阅读此代码。