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

Question

我有一个烧瓶 API，我正在使用 Flask-SQLAlchemy 来处理一个 SQLite 数据库。我有一个存储日志条目的表，我想将最大行数限制为 n。由于插入也是从flask之外的另一个脚本使用原始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

此触发器按预期工作，但我正在努力使用 flask-sqlalchemy 设置它。如何使用 flask-sqlalchemy 设置触发器/执行原始 SQL？ SQL 只需要在创建数据库后执行一次，所以我打算在我的 create_all() 语句之后立即执行它。 我偶然发现了this StackOverflow answer，它提出了一个显然很快就会被弃用的解决方案。我还阅读了SQLAlchemy documentation about custom DDL，但我不知道如何使用flask_sqlalchemy 创建这个自定义DDL。当我像在 SQLAlchemy 文档中那样创建 DDL 时，我收到一条错误消息说 DDL object is not bound to an Engine or Connection.:

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。相关文档为here。

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)

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