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

Question

我有一个烧瓶 API，我正在使用 Flask-SQLAlchemy 来处理 SQLite 数据库。我有一个 table 存储日志条目，我想将最大行数限制为数字 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 Whosebug answer which suggested a solution that is apparently going to be deprecated soon. I also read the 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)

确保解释器在创建表之前已经阅读了这段代码。

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

Create a database trigger in an event listener

sqlite

flask

sqlalchemy

flask-sqlalchemy